Skip to content
Branch: master
Find file Copy path
Find file Copy path
1 contributor

Users who have contributed to this file

231 lines (188 sloc) 7.99 KB
import re
import boto3
import string
import os
import random
import json
dynamodb = boto3.client('dynamodb')
def lambda_handler(event, context):
method = event['httpMethod']
# Get the domain be referenced (either
# or stage will be omitted if the API is behind a
# domain (rather than the api gateway dns)
domain = get_domain(event)
if method == 'GET':
# Called if the user hits the /redirect resource
if event['resource'] == '/redirect':
return api_website(event, domain)
# Called if the user is trying to hit a shortened url
# the pathParameters will contain the http://.../redirect/token_id
# therefore the pathParameters will not be empty
elif event['pathParameters'] is not None:
return retrieve_url(event, domain)
if method == 'POST':
return create_new_url(event, domain)
return {
"statusCode": 200,
"headers": {
"Content-Type": 'text/html',
"Access-Control-Allow-Origin": "*"
"body": "HTTP method not supported."
def create_new_url(event, domain):
post_body = event['body']
url = json.loads(post_body)['destination_url']
# If the user provided a custom token, use that token. Otherwise,
# generate a new token
token = json.loads(post_body)['custom_token'] if 'custom_token' in \
json.loads(post_body) else generate_token()
return_payload = {
"statusCode": 200,
"headers": {
"Content-Type": 'text/html',
"Access-Control-Allow-Origin": "*"
# Validates URL from post_body
if not validate_url(url):
return_payload['body'] = "The provided URL is invalid.\n"
return return_payload
# Put the token and url into DynamoDB
response = dynamodb.put_item(TableName=os.environ['dynamodb_table'],
Item={'id': {'S': "{}".format(token)},
# if the consumer requested a JSON payload in return,
# return json, otherwise just return a string
if 'application/json' in event['headers']['Accept']:
return_payload['headers']['Content-Type'] = 'application/json'
return_payload['body'] = json.dumps({
'shortened_url': '{domain}/{token}'.format(domain=domain,
return_payload['body'] = \
"Shortened URL for {url} created. <br>".format(url=url) + \
"The shortened url is <a href=\"{domain}/{token}\">{domain}/{token}</a><br>".format(domain=domain,
return return_payload
def retrieve_url(event, domain):
return_payload = {
"statusCode": 301,
"headers": {
"Content-Type": 'text/html',
"Access-Control-Allow-Origin": "*"
token = event['pathParameters']['proxy']
# Based on the token, retrieve url from dynamodb table
response = dynamodb.get_item(TableName=os.environ['dynamodb_table'],
Key={'id': {'S': token}})
# if the token key doesn't exist in the dynamodb table, return error body
if 'Item' not in response:
return_payload['statusCode'] = 200
return_payload['body'] = "Token {} Invalid. URL Not Found\n".format(token)
return return_payload
# if the token was found, retrieve the URL from DynamoDB and add it to the
# 'Location' header for the redirect
return_payload['headers']['Location'] = response['Item']['destination_url']['S']
return_payload['body'] = ""
return return_payload
def generate_token():
# Generate a random one time token
allowed_chars = string.digits + string.ascii_letters
return ''.join(random.SystemRandom().choice(allowed_chars)for _ in range(6))
def validate_url(url):
# Validate the given URL
regex = re.compile( r'^(?:http|ftp)s?://' # http:// or https://
r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|' # domain...
r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
r'(?::\d+)?' # optional port
r'(?:/?|[/?]\S+)$', re.IGNORECASE)
if not re.findall(regex, url):
return False
return True
def get_domain(event):
# Supports test invocations from API Gateway
if event['headers'] is None:
return "https://testinvocation/redirect"
# Extracts the domain from event object based on for both api gateway URLs
# or custom domains
if '' in event['headers']['Host']:
return "https://{domain}/{stage}/redirect".format(domain=event['headers']['Host'],
return "https://{domain}/redirect".format(domain=event['headers']['Host'])
def api_website(event, domain):
# returns a website front end for the redirect tool
body = """
<body bgcolor=\"#E6E6FA\">
<!-- Latest compiled and minified CSS -->
<link rel="stylesheet" href="" integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
<script src=""></script>
.form {
padding-left: 1cm;
padding-left: 1cm;
<script src=""></script>
var destinationUrl = document.getElementById("destinationUrl").value;
var dict = {};
dict["destination_url"] = destinationUrl;
if (document.getElementById("customToken").value != "") {
dict["custom_token"] = document.getElementById("customToken").value;
type: 'POST',
headers: {
crossDomain: true,
data: JSON.stringify(dict),
dataType: 'text',
success: function(responseData) {
document.getElementById("id").innerHTML = responseData;
error: function (responseData) {
alert('POST failed.'+ JSON.stringify(responseData));
<title>Serverless URL Redirect</title>
<h1 class="div">Serverless URL Redirect</h1>
<form class="form" action="" method="post">
<textarea rows="1" cols="50" name="text" id="destinationUrl" placeholder="Enter URL ("></textarea>
<form class="form" action="" method="post">
<textarea rows="1" cols="50" name="text" id="customToken" placeholder="Use Custom Token ("></textarea>
<div class="div"><button class="btn btn-primary">Shorten URL</button></div>
<div class="div" id='id'></div>
return {
"statusCode": 200,
"headers": {
"Content-Type": 'text/html',
"Access-Control-Allow-Origin": "*"
"body": string.Template(body).safe_substitute({"domain": domain})
You can’t perform that action at this time.