如何用Lambda借助IAM用户跨账户复制Vendor S3桶文件至我方S3桶？

可行性确认 #

完全可以实现。核心思路是在Lambda函数中初始化两个独立的AWS S3客户端实例：一个使用供应商提供的IAM用户凭证（仅能访问对方S3桶），另一个使用我方Lambda执行角色的权限（或我方IAM用户凭证）来访问自己的S3桶，通过这两个客户端完成文件的读取与上传操作。

具体实现步骤 #

1. 安全存储供应商凭证 #

把供应商提供的IAM用户Access Key ID和Secret Access Key存储在AWS Secrets Manager中（或通过KMS加密Lambda环境变量），绝对不能硬编码到代码里，避免凭证泄露。

2. 初始化双S3客户端 #

在Lambda代码中分别初始化两个S3客户端：

• 供应商S3客户端：使用从Secrets Manager/加密环境变量中获取的供应商凭证创建，用于读取对方桶内的文件。
• 我方S3客户端：直接使用Lambda默认的执行角色权限（需提前给该角色配置我方S3桶的s3:PutObject权限），无需额外凭证，用于上传文件到我方桶。

3. 遍历供应商桶内所有文件 #

通过供应商S3客户端调用list_objects_v2方法遍历桶内对象，注意处理分页逻辑（因为单次调用最多返回1000个对象，需通过NextContinuationToken循环获取所有文件）。

4. 对象传递与上传（推荐流式传输） #

最高效的方式是流式传输，无需将整个文件加载到内存或临时存储，直接把从供应商桶获取的对象流传递给我方S3客户端的上传接口：

• 用供应商客户端的get_object方法获取对象的Body流。
• 把该流直接传入我方客户端的put_object方法的Body参数，完成上传。

如果是小文件，也可以先下载到Lambda的/tmp临时目录再上传，但大文件容易触发内存限制，不推荐。

代码示例（Python + Boto3） #

import boto3
from botocore.exceptions import ClientError

def get_vendor_credentials():
    # 从Secrets Manager获取供应商凭证，替换为你的Secret ARN
    secrets_manager = boto3.client('secretsmanager')
    secret_value = secrets_manager.get_secret_value(SecretId='vendor-s3-credentials')
    return eval(secret_value['SecretString'])  # 假设Secret存储为JSON格式的字典

def lambda_handler(event, context):
    # 初始化供应商S3客户端
    vendor_creds = get_vendor_credentials()
    vendor_s3 = boto3.client(
        's3',
        aws_access_key_id=vendor_creds['access_key'],
        aws_secret_access_key=vendor_creds['secret_key']
    )
    
    # 初始化我方S3客户端（使用Lambda执行角色权限）
    my_s3 = boto3.client('s3')
    
    vendor_bucket = '供应商桶名'
    my_bucket = '我方桶名'
    
    # 遍历供应商桶所有对象
    continuation_token = None
    while True:
        list_kwargs = {'Bucket': vendor_bucket}
        if continuation_token:
            list_kwargs['ContinuationToken'] = continuation_token
        
        try:
            response = vendor_s3.list_objects_v2(**list_kwargs)
        except ClientError as e:
            print(f"遍历供应商桶出错: {e}")
            raise
        
        # 处理当前页的对象
        if 'Contents' in response:
            for obj in response['Contents']:
                obj_key = obj['Key']
                print(f"开始复制文件: {obj_key}")
                
                try:
                    # 获取供应商对象流
                    vendor_obj = vendor_s3.get_object(Bucket=vendor_bucket, Key=obj_key)
                    # 流式上传到我方桶
                    my_s3.put_object(
                        Bucket=my_bucket,
                        Key=obj_key,
                        Body=vendor_obj['Body']
                    )
                    print(f"文件复制完成: {obj_key}")
                except ClientError as e:
                    print(f"复制文件 {obj_key} 出错: {e}")
                    continue
        
        # 检查是否还有下一页
        if response.get('IsTruncated'):
            continuation_token = response['NextContinuationToken']
        else:
            break
    
    return {"status": "success", "message": "所有文件复制完成"}

关键注意事项 #

• 凭证安全：务必用Secrets Manager加密存储供应商凭证，避免明文泄露。
• Lambda配置：根据文件大小调整Lambda内存（内存越高，CPU性能越好，传输速度越快），同时设置足够长的超时时间（比如300秒或更久），避免批量复制时超时。
• 权限验证：提前确认供应商IAM用户拥有s3:GetObject权限，我方Lambda执行角色拥有s3:PutObject权限。
• 错误处理：添加必要的重试逻辑（比如使用tenacity库），处理网络波动、临时权限问题等异常情况。
• 大文件处理：如果涉及GB级大文件，建议改用分段上传（create_multipart_upload），避免内存溢出。

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