more_s3.py

import boto3
import botocore


def key_exists(s3_client, bucket, key):
    """
    Checks if an S3 key exists.
    Args:
        s3_client (S3.Client): The AWS S3 client.
        bucket (str): The AWS S3 bucket name.
        key (str): The AWS S3 key name.
    Returns:
        True if the key exists.
    """
    try:
        # Check the object metadata to see if the key exists
        s3_client.head_object(Bucket=bucket, Key=key)
    except botocore.exceptions.ClientError as e:
        http_status_code = e.response['ResponseMetadata']['HTTPStatusCode']

        if http_status_code == 404:
            # The key does not exist
            return False
        else:
            raise e

    return True


def get_replicated_buckets(s3_client, bucket):
    """
    Gets a list of cross-region buckets replicated from a given bucket.
    Args:
        s3_client (S3.Client): An AWS S3 client.
        bucket (str): An AWS S3 bucket name.
    Returns
        buckets A list of cross-region replicated buckets
    """
    buckets = []

    try:
        response = s3_client.get_bucket_replication(Bucket=bucket)
        rules = response['ReplicationConfiguration']['Rules']

        for rule in rules:
            # The replication rule is ignored if the Status is Disabled
            if rule['Status'] == 'Enabled':
                bucket = rule['Destination']['Bucket'].split(':')[-1]

                buckets.append(bucket)

    except botocore.exceptions.ClientError as e:
        http_status_code = e.response['ResponseMetadata']['HTTPStatusCode']

        if http_status_code == 404:
            # No replication for this bucket
            return buckets
        else:
            raise e

    return buckets


def get_preferred_bucket(s3_client, bucket, region):
    """
    Gets the appropriate bucket based on the region.
    Args:
        s3_client (S3.Client): An AWS S3 client.
        bucket (str): An AWS S3 bucket name.
        region (str): An AWS region name.
    Returns:
        bucket the regional bucket or the original bucket if none exists
    """
    buckets = get_replicated_buckets(s3_client, bucket)

    if not region or not len(buckets):
        # No preference or no bucket replications exists
        return bucket

    for b in buckets:
        response = s3_client.get_bucket_location(Bucket=b)

        if response['LocationConstraint'] == region:
            return b

    return bucket


def is_granted_access(s3_client, bucket, key, grantee):
    """
    Checks if the key can be accessed without a presigned URL.
    Args:
        s3_client (S3.Client): An AWS S3 client.
        bucket (str): An AWS S3 bucket name.
        key (str): An AWS S3 key name.
        grantee (str): An AWS S3 object ACL grantee URI.
    Returns:
        True if the key can be accessed without a presigned URL
    """
    all_users = 'http://acs.amazonaws.com/groups/global/AllUsers'
    acl = s3_client.get_object_acl(Bucket=bucket, Key=key)

    for grant in acl['Grants']:
        if grant['Grantee'].get('URI') == all_users:
            return True

        if grantee and grant['Grantee'].get('URI') == grantee:
            return True

    return False


def get_url(
    s3_client,
    bucket,
    key,
    preferred_region,
    grantee=None,
    expires_in=3600
):
    """
    Gets the content URL based on the preferred region.
    Args:
        s3_client (S3.Client): An AWS S3 client.
        bucket (str): An AWS S3 bucket name.
        region (str): An AWS region name.
        grantee (str): An AWS S3 object ACL grantee URI.
        expires_in (int, optional): Signature expiration seconds.
            Default to an hour.
    Returns:
        url The content URL
    """
    if not key_exists(s3_client, bucket, key):
        # Oops
        return 'return a 404 or something'

    # Get the preffered cross-region replication bucket
    # or the parent bucket if there is no regional preference
    preferred_bucket = get_preferred_bucket(
        s3_client,
        bucket,
        preferred_region
    )

    # Check if the key can be accessed without a presigned URL
    if is_granted_access(s3_client, preferred_bucket, key, grantee):
        response = s3_client.generate_presigned_post(
            Bucket=preferred_bucket,
            Key=key
        )

        return response['url'] + key

    # Get a presigned URL to give to the client
    # in order to access private content
    return s3_client.generate_presigned_url(
        'get_object',
        Params={
            'Bucket': preferred_bucket,
            'Key': key
        },
        ExpiresIn=expires_in
    )


if __name__ == '__main__':
    s3_client = boto3.client('s3')
    # The bucket from which location constrained buckets are replicated
    bucket = 'amber-devops'
    # The assumption is that the key is private, i.e. no public-read ACL
    key = 'blah.html'
    # A client can have a preferred region setting so we can give them
    # a bucket within their preferred region if one exists.
    # This can result in faster download speeds.
    preferred_region = 'us-west-1'

    # Get the presigned URL in the preferred region
    print get_url(s3_client, bucket, key, preferred_region)