如何通过AWS Lambda获取最新创建的S3 Bucket并绑定策略？

Hey there! Let's work through how to get the latest created S3 bucket so your Lambda can automatically attach that policy. I've got two solid approaches for you, depending on your use case:

Approach 1: List All Buckets & Sort by Creation Date #

This works if you need to run the Lambda on a schedule (like hourly) to check for new buckets. It's straightforward, though keep in mind it will scan all your buckets each time.

Here's how to modify your code to fetch the newest bucket:

import boto3
import json

client = boto3.client('s3')

def get_latest_bucket():
    # Fetch all S3 buckets in your account
    all_buckets = client.list_buckets()['Buckets']
    # Sort buckets by creation date (newest first)
    sorted_buckets = sorted(all_buckets, key=lambda bucket: bucket['CreationDate'], reverse=True)
    
    if not sorted_buckets:
        raise Exception("No S3 buckets found in your AWS account")
    
    return sorted_buckets[0]['Name']

def lambda_handler(event, context):
    bucket_name = get_latest_bucket()
    
    # Build your bucket policy with the dynamic bucket name
    bucket_policy = {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "DenyS3PublicObjectACL",
            "Effect": "Deny",
            "Principal": "*",
            "Action": ["s3:PutObjectAcl"],
            "Resource": f"arn:aws:s3:::{bucket_name}/*",
            "Condition": {
                "StringEqualsIgnoreCaseIfExists": {
                    "s3:x-amz-acl": [
                        "public-read",
                        "public-read-write",
                        "authenticated-read"
                    ]
                }
            }
        }]
    }
    
    # Convert policy to JSON string and apply it
    bucket_policy_json = json.dumps(bucket_policy)
    response = client.put_bucket_policy(
        Bucket=bucket_name,
        ConfirmRemoveSelfBucketAccess=True,
        Policy=bucket_policy_json
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps(f"Successfully applied policy to bucket: {bucket_name}")
    }

Approach 2: Trigger Lambda on S3 Bucket Creation (Recommended) #

If you want to attach the policy immediately when a bucket is created, this is the better approach. It avoids polling and is more efficient. You'll need to set up an S3 event notification to trigger your Lambda whenever a new bucket is created.

Here's the modified Lambda code that pulls the bucket name directly from the S3 event:

import boto3
import json

client = boto3.client('s3')

def lambda_handler(event, context):
    # Extract the newly created bucket name from the S3 event
    # Note: The event structure might vary slightly depending on your region/config
    bucket_name = event['detail']['requestParameters']['bucketName']
    
    # Build and apply the policy (same as before, but dynamic to the new bucket)
    bucket_policy = {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "DenyS3PublicObjectACL",
            "Effect": "Deny",
            "Principal": "*",
            "Action": ["s3:PutObjectAcl"],
            "Resource": f"arn:aws:s3:::{bucket_name}/*",
            "Condition": {
                "StringEqualsIgnoreCaseIfExists": {
                    "s3:x-amz-acl": [
                        "public-read",
                        "public-read-write",
                        "authenticated-read"
                    ]
                }
            }
        }]
    }
    
    bucket_policy_json = json.dumps(bucket_policy)
    response = client.put_bucket_policy(
        Bucket=bucket_name,
        ConfirmRemoveSelfBucketAccess=True,
        Policy=bucket_policy_json
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps(f"Successfully applied policy to new bucket: {bucket_name}")
    }

Quick Notes to Keep in Mind: #

• Permissions: Make sure your Lambda execution role has the right permissions:
  • For Approach 1: s3:ListAllMyBuckets and s3:PutBucketPolicy
  • For Approach 2: s3:PutBucketPolicy plus permissions to receive S3 events
• Duplicate Policy Checks: If using Approach 1, you might want to add logic to check if the bucket already has this policy before applying it (to avoid redundant calls)
• Event Structure: For Approach 2, test the Lambda once to verify the event structure if you run into issues extracting the bucket name

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