如何通过boto3验证S3存储桶内所有对象（含旧数据）均已加密？

要确认S3存储桶里的现有对象是否已加密，核心是直接检查每个对象的服务器端加密属性——桶加密策略仅管控未来上传规则，无法反映存量对象的实际加密状态。以下是用Python+Boto3实现的具体方案：

基础遍历检查（非版本化桶） #

对于未开启版本控制的桶，直接遍历所有对象，读取每个对象的ServerSideEncryption元数据：

import boto3

def check_s3_object_encryption(bucket_name):
    s3 = boto3.client('s3')
    unencrypted_objects = []
    continuation_token = None

    while True:
        # 分页获取对象列表（默认单次返回1000条）
        request_params = {'Bucket': bucket_name}
        if continuation_token:
            request_params['ContinuationToken'] = continuation_token
        
        response = s3.list_objects_v2(**request_params)

        # 遍历当前页对象，检查加密状态
        for obj in response.get('Contents', []):
            obj_key = obj['Key']
            obj_meta = s3.head_object(Bucket=bucket_name, Key=obj_key)
            encryption_status = obj_meta.get('ServerSideEncryption')
            
            if not encryption_status:
                unencrypted_objects.append(obj_key)
                print(f"未加密对象: {obj_key}")

        # 终止分页循环
        if not response.get('IsTruncated'):
            break
        continuation_token = response.get('NextContinuationToken')

    print(f"\n检查完成，共发现 {len(unencrypted_objects)} 个未加密对象")
    return unencrypted_objects

# 调用示例
check_s3_object_encryption("your-bucket-name")

版本化桶的检查 #

如果桶开启了版本控制，需要检查每个对象的所有历史版本——旧版本可能未加密：

import boto3

def check_versioned_s3_objects(bucket_name):
    s3 = boto3.client('s3')
    unencrypted_versions = []
    key_marker = None
    version_id_marker = None

    while True:
        # 分页获取所有对象版本
        request_params = {'Bucket': bucket_name}
        if key_marker and version_id_marker:
            request_params['KeyMarker'] = key_marker
            request_params['VersionIdMarker'] = version_id_marker
        
        response = s3.list_object_versions(**request_params)

        # 遍历每个版本，检查加密状态
        for version in response.get('Versions', []):
            obj_key = version['Key']
            version_id = version['VersionId']
            obj_meta = s3.head_object(
                Bucket=bucket_name,
                Key=obj_key,
                VersionId=version_id
            )
            encryption_status = obj_meta.get('ServerSideEncryption')
            
            if not encryption_status:
                unencrypted_versions.append(f"{obj_key} (版本ID: {version_id})")
                print(f"未加密对象版本: {obj_key} - {version_id}")

        # 终止分页循环
        if not response.get('IsTruncated'):
            break
        key_marker = response.get('NextKeyMarker')
        version_id_marker = response.get('NextVersionIdMarker')

    print(f"\n检查完成，共发现 {len(unencrypted_versions)} 个未加密对象版本")
    return unencrypted_versions

# 调用示例
check_versioned_s3_objects("your-versioned-bucket")

关键注意事项 #

• ServerSideEncryption字段存在即代表对象已加密，常见值为AES256（S3托管密钥）或aws:kms（KMS密钥）。
• 针对百万级以上对象的桶，建议用concurrent.futures实现多线程异步检查，大幅提升效率。
• 执行代码的IAM角色需具备s3:ListBucket、s3:GetObject权限，否则会触发权限错误。

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