import boto3 from botocore.exceptions import ClientError import time import uuid import json # A script to create a CloudFront distribution for an S3 bucket using Boto3. # --- Configuration --- REGION = "us-east-1" RANDOM_SUFFIX = uuid.uuid4().hex[:8] S3_BUCKET_NAME = f"my-boto3-cf-s3-origin-{RANDOM_SUFFIX}" CF_DISTRIBUTION_COMMENT = "My Boto3 CloudFront Distribution for S3" OAC_NAME = "MyBoto3CF_OAC" s3_client = boto3.client('s3', region_name=REGION) cloudfront_client = boto3.client('cloudfront', region_name=REGION) sts_client = boto3.client('sts', region_name=REGION) def create_s3_bucket(): """Creates an S3 bucket.""" print(f"--- Creating S3 Bucket: {S3_BUCKET_NAME} ---") try: s3_client.create_bucket( Bucket=S3_BUCKET_NAME, CreateBucketConfiguration={'LocationConstraint': REGION} if REGION != 'us-east-1' else {} ) print("S3 Bucket created.") except ClientError as e: print(f"Error creating S3 bucket: {e}") raise def create_oac(): """Creates an Origin Access Control (OAC).""" print(f"\n--- Creating Origin Access Control (OAC) ---") try: oac_response = cloudfront_client.create_origin_access_control( OriginAccessControlConfig={ 'Name': OAC_NAME, 'SigningBehavior': 'no-override', 'SigningProtocol': 'sigv4', 'OriginAccessControlOriginType': 's3' } ) oac_id = oac_response['OriginAccessControl']['Id'] oac_etag = oac_response['OriginAccessControl']['ETag'] print(f"OAC created with ID: {oac_id}") return oac_id, oac_etag except ClientError as e: print(f"Error creating OAC: {e}") raise def update_s3_bucket_policy(oac_id): """Updates the S3 Bucket Policy to allow OAC access.""" print(f"\n--- Updating S3 Bucket Policy for OAC access ---") try: account_id = sts_client.get_caller_identity()['Account'] # The OAC ARN is not directly available from create_origin_access_control response # It's constructed based on the CloudFront service principal and OAC ID # For S3 bucket policy, we use the CloudFront service principal and SourceArn condition bucket_policy = { "Version": "2012-10-17", "Statement": [ { "Sid": "AllowCloudFrontServicePrincipalReadOnly", "Effect": "Allow", "Principal": { "Service": "cloudfront.amazonaws.com" }, "Action": "s3:GetObject", "Resource": f"arn:aws:s3:::{S3_BUCKET_NAME}/*", "Condition": { "StringEquals": { "AWS:SourceArn": f"arn:aws:cloudfront::{account_id}:distribution/*" } } } ] } s3_client.put_bucket_policy(Bucket=S3_BUCKET_NAME, Policy=json.dumps(bucket_policy)) print("S3 Bucket Policy updated.") except ClientError as e: print(f"Error updating S3 bucket policy: {e}") raise def create_cloudfront_distribution(oac_id): """Creates a CloudFront web distribution.""" print(f"\n--- Creating CloudFront Distribution ---") try: dist_response = cloudfront_client.create_distribution( DistributionConfig={ 'CallerReference': str(uuid.uuid4()), 'Comment': CF_DISTRIBUTION_COMMENT, 'Enabled': True, 'Origins': { 'Quantity': 1, 'Items': [ { 'Id': 'S3Origin', 'DomainName': f"{S3_BUCKET_NAME}.s3.{REGION}.amazonaws.com", 'OriginAccessControlId': oac_id, 'S3OriginConfig': {'OriginAccessIdentity': ''} # OAI is empty when using OAC }, ] }, 'DefaultCacheBehavior': { 'TargetOriginId': 'S3Origin', 'ViewerProtocolPolicy': 'redirect-to-https', 'AllowedMethods': { 'Quantity': 2, 'Items': ['GET', 'HEAD'], 'CachedMethods': {'Quantity': 2, 'Items': ['GET', 'HEAD']} }, 'ForwardedValues': {'QueryString': False, 'Cookies': {'Forward': 'none'}}, 'MinTTL': 0, 'DefaultTTL': 86400, 'MaxTTL': 31536000 }, 'ViewerCertificate': {'CloudFrontDefaultCertificate': True}, 'PriceClass': 'PriceClass_100' } ) dist_id = dist_response['Distribution']['Id'] print(f"CloudFront Distribution created with ID: {dist_id}. Waiting for it to deploy (this can take 10-15 minutes)...") cloudfront_client.get_waiter('distribution_deployed').wait(Id=dist_id) print("CloudFront Distribution deployed.") return dist_id except ClientError as e: print(f"Error creating CloudFront distribution: {e}") raise def cleanup_resources(dist_id, s3_bucket_name, oac_id, oac_etag): """Cleans up all created resources.""" print(f"\n--- Cleaning up resources ---") # Disable and Delete CloudFront Distribution if dist_id: print(f"Disabling CloudFront Distribution '{dist_id}'...") try: dist_config_response = cloudfront_client.get_distribution_config(Id=dist_id) dist_config = dist_config_response['DistributionConfig'] dist_etag = dist_config_response['ETag'] dist_config['Enabled'] = False # Disable the distribution cloudfront_client.update_distribution( DistributionConfig=dist_config, Id=dist_id, IfMatch=dist_etag ) print("Waiting for CloudFront Distribution to be disabled...") cloudfront_client.get_waiter('distribution_deployed').wait(Id=dist_id) print("CloudFront Distribution disabled.") print(f"Deleting CloudFront Distribution '{dist_id}'...") cloudfront_client.delete_distribution(Id=dist_id, IfMatch=dist_etag) print("CloudFront Distribution deleted.") except ClientError as e: if e.response['Error']['Code'] == 'NoSuchDistribution': print(f"CloudFront Distribution '{dist_id}' not found, skipping deletion.") else: print(f"Error deleting CloudFront distribution: {e}") # Delete S3 Bucket if s3_bucket_name: print(f"Deleting S3 Bucket '{s3_bucket_name}'...") try: s3_resource = boto3.resource('s3', region_name=REGION) bucket = s3_resource.Bucket(s3_bucket_name) bucket.objects.all().delete() # Delete all objects in the bucket bucket.delete() print("S3 Bucket deleted.") except ClientError as e: print(f"Error deleting S3 bucket: {e}") # Delete OAC if oac_id and oac_etag: print(f"Deleting OAC '{oac_id}'...") try: cloudfront_client.delete_origin_access_control(Id=oac_id, IfMatch=oac_etag) print("OAC deleted.") except ClientError as e: if e.response['Error']['Code'] == 'NoSuchOriginAccessControl': print(f"OAC '{oac_id}' not found, skipping deletion.") else: print(f"Error deleting OAC: {e}") def main(): dist_id = None oac_id = None oac_etag = None try: create_s3_bucket() oac_id, oac_etag = create_oac() update_s3_bucket_policy(oac_id) dist_id = create_cloudfront_distribution(oac_id) dist_info = cloudfront_client.get_distribution(Id=dist_id) cf_domain_name = dist_info['Distribution']['DomainName'] print("\n--- CloudFront Distribution Setup Complete! ---") print(f"CloudFront Domain Name: {cf_domain_name}") print(f"You can now upload content to s3://{S3_BUCKET_NAME}/ and access it via CloudFront.") input("Press Enter to delete the CloudFront distribution and clean up resources...") except ClientError as e: print(f"An AWS client error occurred: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: cleanup_resources(dist_id, S3_BUCKET_NAME, oac_id, oac_etag) print("\n--- CloudFront demonstration and cleanup complete ---") if __name__ == "__main__": main()