def create_logging_bucket()

in source/remediation_runbooks/scripts/CreateAccessLoggingBucket_createloggingbucket.py [0:0]

• 48 lines of code
• 6 McCabe index (conditional complexity)

def create_logging_bucket(event, context):
    boto_config = Config(
        retries ={
          'mode': 'standard'
        }
    )
    s3 = connect_to_s3(boto_config)

    try:
        kwargs = {
            'Bucket': event['BucketName'],
            'GrantWrite': 'uri=http://acs.amazonaws.com/groups/s3/LogDelivery',
            'GrantReadACP': 'uri=http://acs.amazonaws.com/groups/s3/LogDelivery'
        }
        if event['AWS_REGION'] != 'us-east-1':
            kwargs['CreateBucketConfiguration'] = {
                'LocationConstraint': event['AWS_REGION']
            }

        s3.create_bucket(**kwargs)

        s3.put_bucket_encryption(
            Bucket=event['BucketName'],
            ServerSideEncryptionConfiguration={
                'Rules': [
                    {
                        'ApplyServerSideEncryptionByDefault': {
                            'SSEAlgorithm': 'AES256'
                        }
                    }
                ]
            }
        )
        return {
            "output": {
                "Message": f'Bucket {event["BucketName"]} created'
            }
        }
    except ClientError as error:
        if error.response['Error']['Code'] != 'BucketAlreadyExists' and \
            error.response['Error']['Code'] != 'BucketAlreadyOwnedByYou':
            exit(str(error))
        else:
            return {
                "output": {
                    "Message": f'Bucket {event["BucketName"]} already exists'
                }
            }
    except Exception as e:
        print(e)
        exit(str(e))