跨组织Lambda实现S3传输权限问题排查求助

源组织的CloudWatch运行正常，但Lambda执行s3_assumed.copy_object命令时失败。目标组织的假定角色权限已通过文末测试Lambda验证无问题。

问题出在复制过程中假定身份无法访问源文件。已尝试配置源存储桶策略，但无效；也试过在源账号为目标账号添加信任策略，同样未解决。

源账号配置 #

S3存储桶 #

存储桶ARN：arn:aws:s3:::test-lambda-transfer
已为下方Lambda函数配置PUT和POST事件通知。

策略（部分内容非必需，用于排查权限问题）：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowLambdaGetObject",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::<source-account-id>:role/s3-cross-org-transfer-role"
            },
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::test-lambda-transfer/*"
        },
        {
            "Sid": "AllowLambdaListBucket",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::<source-account-id>:role/s3-cross-org-transfer-role"
            },
            "Action": "s3:ListBucket",
            "Resource": "arn:aws:s3:::test-lambda-transfer"
        },
        {
            "Sid": "AllowAssumedRoleGetObject",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::<destination-account-id>:role/test-cross-org-transfer-role"
            },
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::test-lambda-transfer/*"
        }
    ]
}

Lambda函数 #

Lambda ARN：arn:aws:lambda:us-east-1::<source-account-id>:function:cross-account-s3-transfer
权限角色：arn:aws:iam::<source-account-id>:role/s3-cross-org-transfer-role
角色策略：arn:aws:iam::<source-account-id>:policy/s3-transfer-execution-policy

import boto3
import urllib.parse
import logging

# 初始化日志
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    sts_client = boto3.client('sts')
    destination_bucket = 's3-transfer-test-receiver-bucket'
    role_to_assume_arn = 'arn:aws:iam::<destination-account-id>:role/test-cross-org-transfer-role'

    try:
        for record in event['Records']:
            source_bucket = record['s3']['bucket']['name']
            key = urllib.parse.unquote_plus(record['s3']['object']['key'])
            logger.info(f"检测到新对象: {key}，位于存储桶 {source_bucket}")

            # 尝试从源存储桶获取对象元数据
            s3_source = boto3.client('s3')
            try:
                response = s3_source.head_object(Bucket=source_bucket, Key=key)
                logger.info(f"成功获取 {key} 在 {source_bucket} 的元数据")
            except Exception as e:
                logger.error(f"获取 {key} 在 {source_bucket} 的元数据失败: {str(e)}")
                raise e

            logger.info("正在扮演目标角色...")
            assumed_role = sts_client.assume_role(RoleArn=role_to_assume_arn, RoleSessionName="S3CopySession")
            logger.info(f"已扮演角色ARN: {assumed_role['AssumedRoleUser']['Arn']}")

            credentials = assumed_role['Credentials']
            s3_assumed = boto3.client('s3',
                                      aws_access_key_id=credentials['AccessKeyId'],
                                      aws_secret_access_key=credentials['SecretAccessKey'],
                                      aws_session_token=credentials['SessionToken'])

            copy_source = {'Bucket': source_bucket, 'Key': key}
            logger.info(f"正在复制 {key} 从 {source_bucket} 到 {destination_bucket}")
            s3_assumed.copy_object(Bucket=destination_bucket, Key=key, CopySource=copy_source)
            logger.info(f"成功复制 {key} 到 {destination_bucket}")

    except Exception as e:
        logger.error(f"复制操作失败: {str(e)}")
        raise e

Lambda函数策略

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::test-lambda-transfer",
                "arn:aws:s3:::test-lambda-transfer/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": "sts:AssumeRole",
            "Resource": "arn:aws:iam::<destination-account-id>:role/test-cross-org-transfer-role"
        }
    ]
}

目标账号配置 #

S3存储桶 #

存储桶ARN：arn:aws:s3:::s3-transfer-test-receiver-bucket

IAM配置 #

角色：arn:aws:iam::<destination-account-id>:role/test-cross-org-transfer-role
策略：arn:aws:iam::<destination-account-id>:policy/put-object-s3-cross-account-test-policy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::s3-transfer-test-receiver-bucket/*"
        }
    ]
}

信任关系 - 可信实体：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        },
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": [
                    "arn:aws:iam::<source-account-id>:role/s3-cross-org-transfer-role",
                    "arn:aws:iam::<destination-account-id>:role/test-cross-org-transfer-role"
                ]
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

测试 #

可正常运行的Python Lambda测试代码 #

import boto3
import logging

# 配置日志
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    try:
        sts_client = boto3.client('sts')
        logger.info("正在扮演目标角色...")
        assumed_role = sts_client.assume_role(
            RoleArn="arn:aws:iam::<destination-account-id>:role/test-cross-org-transfer-role",
            RoleSessionName="CrossAccountSession"
        )
        logger.info(f"已扮演角色ARN: {assumed_role['AssumedRoleUser']['Arn']}")
        logger.info("角色扮演成功")

        credentials = assumed_role['Credentials']
        s3 = boto3.client(
            's3',
            aws_access_key_id=credentials['AccessKeyId'],
            aws_secret_access_key=credentials['SecretAccessKey'],
            aws_session_token=credentials['SessionToken'],
        )

        destination_bucket = 's3-transfer-test-receiver-bucket'

        # 尝试直接上传一个测试对象
        test_key = 'test-object.txt'
        test_body = '这是一个测试对象。'
        logger.info(f"正在直接上传测试对象到 {destination_bucket}...")
        s3.put_object(Bucket=destination_bucket, Key=test_key, Body=test_body)
        logger.info("直接上传成功")

    except Exception as e:
        logger.error(f"直接上传操作失败: {e}")
        raise e

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