Script to create redirects in S3 buckets

README.md

s3redirect

Authentication

Your AWS credentials can be set in a number of ways:

1. In a ".boto" file in your home folder. See Boto's documentation for how to create this file.
2. In the environment variables "AWS_ACCESS_KEY_ID" and "AWS_SECRET_ACCESS_KEY".
3. Passed in as arguments. -a or --access-key for the Access Key ID and -s or --access-secret for the Secret Access Key.

s3redirect.py

#!/usr/bin/env python

import os, sys, argparse
from urlparse import urlparse

def alert(text):
    sys.stdout.write(text + '\n')
    sys.stdout.flush()
    
try:
    import boto
except ImportError:
    alert("Please install boto. `pip install boto`")
    sys.exit(os.EX_UNAVAILABLE)
    
try:
    import xmltodict
except ImportError:
    alert("Please install xmltodict. `pip install xmltodict`")
    sys.exit(os.EX_UNAVAILABLE)
    
parser          =   argparse.ArgumentParser()

parser.add_argument("bucket", help = "Which bucket to put the redirect in", type = str)
parser.add_argument("key", help = "The key to redirect", type = str)
parser.add_argument("uri", help = "The URI to redirect to", type = str)

parser.add_argument('-a', '--access-key', help = "AWS Access Key ID", type = str)
parser.add_argument('-s', '--access-secret', help = "AWS Access Key Secret", type = str)
parser.add_argument('-u', '--update', help = "Update an existing S3 key", action = "store_true", default = False)

args            =   parser.parse_args()

AWS_KEY         =   args.access_key
AWS_SECRET      =   args.access_secret

# lookup global AWS keys if needed
if AWS_KEY is None:
    AWS_KEY     =   boto.config.get('Credentials', 'aws_access_key_id')
    
if AWS_SECRET is None:
    AWS_SECRET  =   boto.config.get('Credentials', 'aws_secret_access_key')
    
# lookup AWS key environment variables
if AWS_KEY is None:
    AWS_KEY     =   os.environ.get('AWS_ACCESS_KEY_ID')
if AWS_SECRET is None:
    AWS_SECRET  =   os.environ.get('AWS_SECRET_ACCESS_KEY')

s3connection        =   boto.connect_s3(AWS_KEY, AWS_SECRET)

# test the bucket connection
try:
    s3bucket        =   s3connection.get_bucket(args.bucket)
except boto.exception.S3ResponseError:
    alert('Bucket "%s" could not be retrieved with the specified credentials' % args.bucket)
    sys.exit(os.EX_NOINPUT)
    
# check for an existing key first since that's cheaper
s3key           =   s3bucket.get_key(args.key)
if not s3key is None and not args.update:
    alert('Key "%s" is already in use. Please use the --update option to use this key anyway.' % args.key)
    sys.exit(os.EX_UNAVAILABLE)
    
try:
    config          =   xmltodict.parse(s3bucket.get_website_configuration_with_xml()[1])
except boto.exception.S3ResponseError:
    config          =   False
if config: # if this bucket is a website, use the RedirectRules to handle the redirect
    suffix          =   config.get('WebsiteConfiguration').get('IndexDocument').get('Suffix')
    error_key       =   config.get('WebsiteConfiguration').get('ErrorDocument').get('Key')
    rules           =   boto.s3.website.RoutingRules()
    for route in config.get('WebsiteConfiguration').get('RoutingRules').get('RoutingRule'):
        prefix      =   route.get('Condition').get('KeyPrefixEquals')
        if prefix == args.key and not args.update:
            alert('There is already a redirect for "%s". Please use the --update option to use this key anyway.' % args.key)
            sys.exit(os.EX_UNAVAILABLE)
        elif not prefix == args.key:
            rules.add_rule(boto.s3.website.RoutingRule.when(key_prefix = prefix).then_redirect(hostname = route.get('Redirect').get('HostName'), protocol = route.get('Redirect').get('Protocol'), replace_key = route.get('Redirect').get('ReplaceKeyWith')))
    
    uri             =   urlparse(args.uri)
    path            =   uri.path or None
    if uri.query:
        path        =   path + '?' + uri.query
    rules.add_rule(boto.s3.website.RoutingRule.when(key_prefix = args.key).then_redirect(hostname = uri.netloc or None, protocol = uri.scheme or None, replace_key = path.lstrip('/')))

    if s3bucket.configure_website(suffix = suffix, error_key = error_key, routing_rules = rules):
        alert('%s/%s has been redirected to %s' % (args.bucket, args.key.lstrip('/'), args.uri))
        sys.exit(os.EX_OK)
    else:
        alert('There was an error updating the redirect rules.')
        sys.exit(os.EX_UNAVAILABLE)
else: # This bucket isn't a website so we'll have to make a key
    s3key           =   s3bucket.new_key(args.key)
    s3key.set_redirect(args.uri)
    s3key.set_acl('public-read')

    alert('%s/%s has been redirected to %s' % (args.bucket, args.key.lstrip('/'), args.uri))
    sys.exit(os.EX_OK)