boto3 S3分片上传SHA1校验相关技术疑问及示例需求

核心问题解答 #

1. 响应中的SHA1是否由AWS计算？不匹配会怎样？ #

• 响应里的ChecksumSHA1是AWS基于接收到的分片原始内容计算的结果。
• 当你上传时传入自行计算的ChecksumSHA1，S3会自动执行校验：
  • 若两者不匹配，S3直接返回400 Bad Request错误（错误码InvalidChecksum），该分片不会被存储。
  • 只有校验匹配时，才会返回200响应并附带S3计算的SHA1值。

2. 500MiB分片的SHA1，本地计算与AWS返回值是否会不一致？ #

分片大小超过16MB时Etag不等于MD5的逻辑，和SHA1无关。只要你计算SHA1时用的是未经过任何修改的分片原始数据，本地计算的SHA1就会和AWS返回的完全一致，和分片大小没有关系。

3. 分片上传的完整性验证方案与代码示例 #

分片SHA1对比方式

调用upload_part()后，直接将本地计算的chunk_sha1（注意要和S3返回的Base64编码格式一致）与响应中的ChecksumSHA1做字符串对比。如果不一致，说明传输中数据损坏，需重新上传该分片。

完整文件完整性验证

S3不会自动计算分片上传对象的整体SHA1，推荐两种验证方式：

• 方式一：下载后校验：本地预计算整个文件的SHA1，上传完成后下载文件重新计算对比，适合小文件。
• 方式二：聚合分片SHA1（推荐）：
  1. 计算每个分片的SHA1（二进制格式），将所有分片的SHA1二进制数据拼接。
  2. 对拼接后的二进制数据再计算一次SHA1，得到整个文件的聚合SHA1。
  3. 把聚合SHA1存储到对象标签或元数据中，后续验证时重新计算对比即可。

boto3分片上传+完整性校验示例

import boto3
import hashlib
import base64

def calculate_sha1(data):
    """计算数据的SHA1并返回Base64编码字符串"""
    sha1_hash = hashlib.sha1()
    sha1_hash.update(data)
    return base64.b64encode(sha1_hash.digest()).decode('utf-8')

def upload_part_with_checksum(s3_client, chunk, bucket, key, part_number, upload_id):
    # 计算当前分片的SHA1
    local_sha1 = calculate_sha1(chunk)
    try:
        resp = s3_client.upload_part(
            Bucket=bucket,
            Key=key,
            PartNumber=part_number,
            UploadId=upload_id,
            Body=chunk,
            ChecksumAlgorithm='SHA1',
            ChecksumSHA1=local_sha1
        )
        # 对比本地与S3返回的SHA1
        if resp['ChecksumSHA1'] != local_sha1:
            raise Exception(f"分片{part_number} SHA1校验失败")
        return {
            'PartNumber': part_number,
            'ETag': resp['ETag'],
            'ChecksumSHA1': resp['ChecksumSHA1']
        }
    except s3_client.exceptions.InvalidChecksum:
        raise Exception(f"S3验证分片{part_number} SHA1不匹配，上传失败")

def multipart_upload_with_integrity(s3_client, file_path, bucket, key, chunk_size=500*1024*1024):
    # 初始化分片上传
    upload_resp = s3_client.create_multipart_upload(
        Bucket=bucket,
        Key=key,
        ChecksumAlgorithm='SHA1'
    )
    upload_id = upload_resp['UploadId']
    parts = []
    chunk_sha1s = []

    try:
        with open(file_path, 'rb') as f:
            part_number = 1
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                # 上传分片并校验
                part = upload_part_with_checksum(s3_client, chunk, bucket, key, part_number, upload_id)
                parts.append(part)
                # 保存分片二进制SHA1用于聚合计算
                chunk_sha1_bin = base64.b64decode(part['ChecksumSHA1'])
                chunk_sha1s.append(chunk_sha1_bin)
                part_number += 1

        # 完成分片上传
        complete_resp = s3_client.complete_multipart_upload(
            Bucket=bucket,
            Key=key,
            UploadId=upload_id,
            MultipartUpload={'Parts': parts}
        )

        # 计算整个文件的聚合SHA1
        aggregate_sha1 = hashlib.sha1(b''.join(chunk_sha1s)).digest()
        aggregate_sha1_b64 = base64.b64encode(aggregate_sha1).decode('utf-8')

        # 将聚合SHA1存储到对象标签
        s3_client.put_object_tagging(
            Bucket=bucket,
            Key=key,
            Tagging={
                'TagSet': [
                    {'Key': 'AggregateSHA1', 'Value': aggregate_sha1_b64}
                ]
            }
        )

        print(f"上传完成，聚合SHA1: {aggregate_sha1_b64}")
        return complete_resp

    except Exception as e:
        # 上传失败，中止分片上传
        s3_client.abort_multipart_upload(
            Bucket=bucket,
            Key=key,
            UploadId=upload_id
        )
        raise e

# 使用示例
if __name__ == "__main__":
    s3 = boto3.client('s3')
    file_path = '/path/to/your/large/file'
    bucket_name = 'your-bucket-name'
    object_key = 'your-object-key'
    multipart_upload_with_integrity(s3, file_path, bucket_name, object_key)

关于GetObjectProperties返回的ChecksumSHA1

分片上传对象通过GetObjectProperties返回的ChecksumSHA1并非整个对象的SHA1，而是上传时提供的第一个分片的SHA1（S3当前实现逻辑），不能用于验证整个文件完整性，需使用自行计算的聚合SHA1或重新下载校验。

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