使用boto3上传大文件并实现ContentMD5校验遇到的问题

我之前也踩过这个S3上传校验的坑，尤其是KMS加密桶的情况，给你梳理下问题根源和可行的解决方案：

先理清楚你遇到的所有问题点 #

• upload_file确实是AWS封装的多部分上传工具，但它把ContentMD5参数藏起来了，没法直接用来做校验
• 启用KMS加密的桶里，S3的ETag会变成分块MD5的拼接值（因为加密是按分块处理的），所以本地算的文件MD5和S3返回的ETag完全对不上，这条路走不通
• Object.put虽然支持ContentMD5，但只能单流传小文件，大文件用这个要么超时要么直接报错

可行的解决方案：手动实现多部分上传，分块校验+整体MD5留存 #

S3的多部分上传本身不支持对整个文件做ContentMD5的服务器端验证，但我们可以用两步来保证上传的完整性：

1. 给每个上传分块计算ContentMD5，在upload_part时传进去，让S3帮我们验证每一块的完整性
2. 自己算整个文件的MD5，把它存在文件的Metadata里，之后下载或者需要校验的时候，拿出来和本地/下载后的文件MD5对比就行

直接上可用的代码示例 #

import hashlib
import base64
import boto3
from botocore.exceptions import ClientError

def calc_file_md5(file_path, chunk_size=8*1024*1024):
    """专门给大文件算MD5的方法，分块读取不占内存"""
    md5_hash = hashlib.md5()
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            md5_hash.update(chunk)
    return md5_hash.digest()

def calc_chunk_md5(chunk_data):
    """计算单个上传分块的MD5，转成Base64格式给S3用"""
    md5 = hashlib.md5(chunk_data)
    return base64.b64encode(md5.digest()).decode('utf-8')

def upload_large_file_with_md5_check(bucket_name, local_file, s3_key, kms_key_id, chunk_size=8*1024*1024):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)

    # 第一步：初始化多部分上传
    try:
        multipart_upload = bucket.initiate_multipart_upload(
            Key=s3_key,
            ServerSideEncryption='aws:kms',
            SSEKMSKeyId=kms_key_id
        )
    except ClientError as e:
        print(f"初始化多部分上传失败：{e.response['Error']['Message']}")
        return False

    parts = []
    part_num = 1
    total_file_md5 = hashlib.md5()

    try:
        with open(local_file, 'rb') as f:
            while chunk := f.read(chunk_size):
                # 算当前分块的MD5，传给S3做校验
                chunk_md5 = calc_chunk_md5(chunk)
                # 上传这个分块
                uploaded_part = multipart_upload.upload_part(
                    Body=chunk,
                    PartNumber=part_num,
                    ContentMD5=chunk_md5
                )
                parts.append({
                    'PartNumber': part_num,
                    'ETag': uploaded_part['ETag']
                })
                # 更新整个文件的MD5
                total_file_md5.update(chunk)
                part_num += 1

        # 把整个文件的MD5转成Base64，存在Metadata里
        full_file_md5_b64 = base64.b64encode(total_file_md5.digest()).decode('utf-8')
        # 完成多部分上传，同时把整体MD5写进Metadata
        multipart_upload.complete_multipart_upload(
            MultipartUpload={'Parts': parts},
            Metadata={'full_file_md5': full_file_md5_b64}
        )
        print(f"文件上传成功！整体MD5已存入Metadata：{full_file_md5_b64}")
        return True
    except ClientError as e:
        print(f"上传过程出错：{e.response['Error']['Message']}")
        # 出错一定要中止多部分上传，不然会产生冗余的碎片
        multipart_upload.abort_multipart_upload()
        return False
    except Exception as e:
        print(f"未知错误：{str(e)}")
        multipart_upload.abort_multipart_upload()
        return False

# 调用示例，替换成你的参数就行
if __name__ == "__main__":
    MY_BUCKET = "your-bucket-name"
    LOCAL_FILE_PATH = "path/to/your/large-file.bin"
    S3_OBJECT_KEY = "storage/large-file.bin"
    KMS_KEY_ARN = "your-kms-key-arn-or-id"

    upload_large_file_with_md5_check(MY_BUCKET, LOCAL_FILE_PATH, S3_OBJECT_KEY, KMS_KEY_ARN)

几个关键细节要注意 #

• 分块ContentMD5的作用：S3收到分块后会自动校验这个值，如果分块在传输中损坏，会直接返回错误，保证每一块都传对了
• 整体MD5的作用：虽然S3不会帮我们验证整个文件的MD5，但我们自己存在Metadata里后，后续可以通过obj = s3.Object(bucket, key); md5_from_s3 = obj.metadata['full_file_md5']拿到这个值，和下载后的文件MD5对比，确保整个文件完整
• 分块大小选择：建议用8MB或16MB，这是S3推荐的大小，既能避免请求过多，也不会让单个分块太大导致传输超时

额外提个替代方案（如果用新版SDK） #

如果你用的是boto3 1.26以上的版本，可以试试S3的ChecksumAlgorithm参数，支持SHA-256这类校验算法，S3会自动计算并存储校验和，下载的时候可以直接验证。虽然这不是ContentMD5，但同样能解决完整性校验的问题，用法会更简单一些。

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