def configure_s3_bucket()

in source/custom_resource/custom-resource.py [0:0]

• 45 lines of code
• 7 McCabe index (conditional complexity)

def configure_s3_bucket(log, region, bucket_name, access_logging_bucket_name):
    log.info("[configure_s3_bucket] Start")

    if bucket_name.strip() == "":
        raise Exception('Failed to configure access log bucket. Name cannot be empty!')

    # ------------------------------------------------------------------------------------------------------------------
    # Create the S3 bucket (if not exist)
    # ------------------------------------------------------------------------------------------------------------------
    s3_client = create_client('s3')

    try:
        response = s3_client.head_bucket(Bucket=bucket_name)
        log.info("[configure_s3_bucket]response head_bucket: \n%s" % response)

        # Enable access logging if needed
        put_s3_bucket_access_logging(log, s3_client, bucket_name, access_logging_bucket_name)
    except botocore.exceptions.ClientError as e:
        # If a client error is thrown, then check that it was a 404 error.
        # If it was a 404 error, then the bucket does not exist.
        error_code = int(e.response['Error']['Code'])
        if error_code == 404:
            log.info("[configure_s3_bucket]: %s doesn't exist. Create bucket." % bucket_name)
            if region == 'us-east-1':
                s3_client.create_bucket(Bucket=bucket_name, ACL='private')
            else:
                s3_client.create_bucket(Bucket=bucket_name, ACL='private',
                                        CreateBucketConfiguration={'LocationConstraint': region})

            # Begin waiting for the S3 bucket, mybucket, to exist
            s3_bucket_exists_waiter = s3_client.get_waiter('bucket_exists')
            s3_bucket_exists_waiter.wait(Bucket=bucket_name)

            # Enable server side encryption on the S3 bucket
            response = s3_client.put_bucket_encryption(
                Bucket=bucket_name,
                ServerSideEncryptionConfiguration={
                    'Rules': [
                        {
                            'ApplyServerSideEncryptionByDefault': {
                                'SSEAlgorithm': 'AES256'
                            }
                        },
                    ]
                }
            )
            log.info("[configure_s3_bucket]response put_bucket_encryption: \n%s" % response)
            
            # block public access
            response = s3_client.put_public_access_block(
                Bucket=bucket_name,
                PublicAccessBlockConfiguration={
                    'BlockPublicAcls': True,
                    'IgnorePublicAcls': True,
                    'BlockPublicPolicy': True,
                    'RestrictPublicBuckets': True
                }
            )
            log.info("[configure_s3_bucket]response put_public_access_block: \n%s" % response)

            # Enable access logging
            put_s3_bucket_access_logging(log, s3_client, bucket_name, access_logging_bucket_name)

    log.info("[configure_s3_bucket] End")