使用不同CMK KMS密钥跨S3桶复制加密内容的方案问询

你的代码报错是因为KMS的decrypt接口仅用于解密数据密钥（DEK），而非直接解密整个文件内容，该接口对传入的CiphertextBlob有最大6KB的限制。客户端加密采用信封加密逻辑：实际文件由数据密钥（DEK）加密，DEK再用CMK加密后与文件内容一同存储在S3中。你直接把整个加密文件传给KMS，自然触发长度限制报错。

Python中对应JavaAmazonS3EncryptionClient的是boto3的S3加密客户端，它会自动处理信封加密的解密、重新加密流程，无需手动调用KMS接口处理文件内容。

依赖安装 #

先安装所需依赖：

pip install boto3 aws-encryption-sdk

完整代码示例 #

import boto3
from aws_encryption_sdk import S3EncryptionClient
from aws_encryption_sdk.identifiers import CommitmentPolicy
from aws_encryption_sdk.key_providers.kms import KMSMasterKeyProvider

# 配置参数
SOURCE_BUCKET = "sourceBucket"
DESTINATION_BUCKET = "destinationBucket"
SOURCE_KMS_KEY_ID = "kmsKey"
DEST_KMS_KEY_ID = "destKmsKeyId"
REGION = "us-west-2"

# 初始化KMS主密钥提供者：用于解密源文件
source_kms_provider = KMSMasterKeyProvider(key_ids=[SOURCE_KMS_KEY_ID], region=REGION)
# 初始化目标KMS主密钥提供者：用于加密目标文件
dest_kms_provider = KMSMasterKeyProvider(key_ids=[DEST_KMS_KEY_ID], region=REGION)

# 初始化S3加密客户端，设置承诺策略
encryption_client = S3EncryptionClient(commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT)

# 初始化普通S3客户端用于列出对象
s3_client = boto3.client("s3", region_name=REGION)

def copy_encrypted_objects():
    # 分页列出源桶对象，避免单次返回结果不全
    paginator = s3_client.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=SOURCE_BUCKET):
        if "Contents" not in page:
            continue
        for obj in page["Contents"]:
            file_key = obj["Key"]
            print(f"Processing object: {file_key}")
            
            # 从源桶解密获取对象内容
            with encryption_client.decrypt_object(
                s3_client=s3_client,
                bucket=SOURCE_BUCKET,
                key=file_key,
                master_key_provider=source_kms_provider
            ) as decrypted_object:
                decrypted_content = decrypted_object.read()
            
            # 将解密内容用目标KMS密钥加密后存入目标桶
            encryption_client.put_object(
                s3_client=s3_client,
                bucket=DESTINATION_BUCKET,
                key=file_key,
                body=decrypted_content,
                master_key_provider=dest_kms_provider
            )
            print(f"Successfully copied {file_key} to destination bucket")

if __name__ == "__main__":
    copy_encrypted_objects()

关键说明 #

• 信封加密自动处理：S3EncryptionClient会自动解析源文件中的加密DEK，调用KMS用源CMK解密DEK后解密文件内容；随后生成新DEK，用目标CMK加密DEK，再用新DEK加密文件内容并存入目标桶。
• 分页处理：使用分页器paginator适配源桶中大量对象的场景，避免单次列表请求返回结果不全。
• 权限要求：确保执行代码的IAM身份拥有以下权限：
  • 源S3桶的s3:GetObject权限
  • 目标S3桶的s3:PutObject权限
  • 源CMK的kms:Decrypt权限
  • 目标CMK的kms:GenerateDataKey和kms:Encrypt权限

如果处理超大文件，可改用流式读取/写入避免占用过多内存，示例片段：

# 流式解密并上传
with encryption_client.decrypt_object(
    s3_client=s3_client,
    bucket=SOURCE_BUCKET,
    key=file_key,
    master_key_provider=source_kms_provider
) as decrypted_object:
    encryption_client.put_object(
        s3_client=s3_client,
        bucket=DESTINATION_BUCKET,
        key=file_key,
        body=decrypted_object,
        master_key_provider=dest_kms_provider
    )

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