如何改造Python AWS Lambda实现新建S3桶自动开启默认加密

功能要求 #

• 新建S3存储桶时自动触发Lambda函数运行
• 若存储桶未开启默认加密则自动完成配置
• 触发SNS主题向管理员及存储桶创建者/所有者发送邮件通知

前置配置说明 #

你现有的代码是定时全量扫描所有存量存储桶的逻辑，要适配新建桶触发的场景，需要先完成以下云资源配置：

1. 开通CloudTrail跟踪，开启S3桶创建事件的日志采集
2. 创建EventBridge规则，匹配CreateBucket API调用事件，将触发目标设置为你的Lambda函数
3. 给Lambda执行角色添加以下权限：

• S3的GetBucketEncryption、PutBucketEncryption权限
• SNS的Publish权限
• Lambda基础执行权限

修改后完整Lambda代码 #

import json
import boto3

s3_client = boto3.client("s3")
sns_client = boto3.client("sns")
# 替换为你实际的SNS主题ARN
SNS_TOPIC_ARN = "arn:aws:sns:区域:账号ID:主题名"

def lambda_handler(event, context):
    # 从EventBridge事件中提取新建桶的名称和创建者信息
    bucket_name = event["detail"]["requestParameters"]["bucketName"]
    creator_arn = event["detail"]["userIdentity"]["arn"]
    
    status = 200
    operation_result = ""
    notify_content = ""
    
    # 检查桶是否已开启默认加密
    try:
        s3_client.get_bucket_encryption(Bucket=bucket_name)
        operation_result = f"存储桶{bucket_name}已开启默认加密，无需操作"
        print(operation_result)
        notify_content = f"""
        存储桶创建检测通知：
        存储桶名称：{bucket_name}
        状态：已默认开启加密，无需额外操作
        创建者ARN：{creator_arn}
        """
    except s3_client.exceptions.ClientError:
        # 未开启加密，执行加密配置
        try:
            s3_client.put_bucket_encryption(
                Bucket=bucket_name,
                ServerSideEncryptionConfiguration={
                    'Rules': [
                        {
                            'ApplyServerSideEncryptionByDefault':
                                {
                                    'SSEAlgorithm': 'AES256'
                                }
                        }
                    ]
                }
            )
            operation_result = f"存储桶{bucket_name}未开启加密，已自动配置AES256默认加密"
            print(operation_result)
            notify_content = f"""
            存储桶自动加密操作通知：
            存储桶名称：{bucket_name}
            状态：未开启默认加密，已自动完成AES256加密配置
            创建者ARN：{creator_arn}
            """
        except s3_client.exceptions.ClientError as e:
            status = 500
            operation_result = f"存储桶{bucket_name}加密配置失败，错误信息：{str(e)}"
            print(operation_result)
            notify_content = f"""
            存储桶加密操作失败通知：
            存储桶名称：{bucket_name}
            状态：加密配置失败
            错误信息：{str(e)}
            创建者ARN：{creator_arn}
            """
    
    # 发送SNS通知
    try:
        sns_client.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject=f"S3存储桶加密检测通知",
            Message=notify_content
        )
    except Exception as e:
        print(f"SNS通知发送失败：{str(e)}")
    
    return {
        'statusCode': status,
        'operationResult': operation_result,
        'bucketName': bucket_name
    }

使用说明 #

将代码中的SNS_TOPIC_ARN替换为你实际的SNS资源参数即可使用，原有全量扫描存量桶的逻辑可以单独保留做定期执行，两个逻辑互不影响。

内容的提问来源于stack exchange，提问作者Ruwan Vimukthi Mettananda