Creating Signed URLs for Amazon CloudFront
Amazon CloudFront Signed URLs work differently than Amazon S3 signed URLs. CloudFront uses RSA signatures based on a separate CloudFront keypair which you have to set up in your Amazon Account Credentials page. Here's some code to actually generate a time-limited URL in Python using the M2Crypto library:
Create a keypair for CloudFront
I think the only way to do this is through Amazon's web site. Go into your AWS "Account" page and click on the "Security Credentials" link. Click on the "Key Pairs" tab then click "Create a New Key Pair". This will generate a new key pair for you and automatically download a private key file (pk-xxxxxxxxx.pem). Keep the key file safe and private. Also note down the "Key Pair ID" from amazon as we will need it in the next step.
Generate some URLs in Python
As of boto version 2.0 there does not seem to be any support for generating signed CloudFront URLs. Python does not include RSA encryption routines in the standard library so we will have to use an additional library. I've used M2Crypto in this example.
For a non-streaming distribution, you must use the full cloudfront URL as the resource, however for streaming we only use the object name of the video file. See the code below for a full example of generating a URL which only lasts for 5 minutes.
This code is based loosely on the PHP example code provided by Amazon in the CloudFront documentation.
from M2Crypto import EVP
import base64
import time
def aws_url_base64_encode(msg):
msg_base64 = base64.b64encode(msg)
msg_base64 = msg_base64.replace('+', '-')
msg_base64 = msg_base64.replace('=', '_')
msg_base64 = msg_base64.replace('/', '~')
return msg_base64
def sign_string(message, priv_key_string):
key = EVP.load_key_string(priv_key_string)
key.reset_context(md='sha1')
key.sign_init()
key.sign_update(message)
signature = key.sign_final()
return signature
def create_url(url, encoded_signature, key_pair_id, expires):
signed_url = "%(url)s?Expires=%(expires)s&Signature=%(encoded_signature)s&Key-Pair-Id=%(key_pair_id)s" % {
'url':url,
'expires':expires,
'encoded_signature':encoded_signature,
'key_pair_id':key_pair_id,
}
return signed_url
def get_canned_policy_url(url, priv_key_string, key_pair_id, expires):
#we manually construct this policy string to ensure formatting matches signature
canned_policy = '{"Statement":[{"Resource":"%(url)s","Condition":{"DateLessThan":{"AWS:EpochTime":%(expires)s}}}]}' % {'url':url, 'expires':expires}
#sign the non-encoded policy
signature = sign_string(canned_policy, priv_key_string)
#now base64 encode the signature (URL safe as well)
encoded_signature = aws_url_base64_encode(signature)
#combine these into a full url
signed_url = create_url(url, encoded_signature, key_pair_id, expires);
return signed_url
def encode_query_param(resource):
enc = resource
enc = enc.replace('?', '%3F')
enc = enc.replace('=', '%3D')
enc = enc.replace('&', '%26')
return enc
#Set parameters for URL
key_pair_id = "APKAIAZVIO4BQ" #from the AWS accounts CloudFront tab
priv_key_file = "cloudfront-pk.pem" #your private keypair file
# Use the FULL URL for non-streaming:
resource = "http://34254534.cloudfront.net/video.mp4"
#resource = 'video.mp4' #your resource (just object name for streaming videos)
expires = int(time.time()) + 300 #5 min
#Create the signed URL
priv_key_string = open(priv_key_file).read()
signed_url = get_canned_policy_url(resource, priv_key_string, key_pair_id, expires)
print(signed_url)
#Flash player doesn't like query params so encode them if you're using a streaming distribution
#enc_url = encode_query_param(signed_url)
#print(enc_url)
Make sure that you set up your distribution with a TrustedSigners parameter set to the account holding your keypair (or "Self" if it's your own account)
See Getting started with secure AWS CloudFront streaming with Python for a fully worked example on setting this up for streaming with Python
This feature is now already supported in Botocore, which is the underlying library of Boto3, the latest official AWS SDK for Python. (The following sample requires the installation of the rsa package, but you can use other RSA package too, just define your own "normalized RSA signer".)
The usage looks like this:
from botocore.signers import CloudFrontSigner
# First you create a cloudfront signer based on a normalized RSA signer::
import rsa
def rsa_signer(message):
private_key = open('private_key.pem', 'r').read()
return rsa.sign(
message,
rsa.PrivateKey.load_pkcs1(private_key.encode('utf8')),
'SHA-1') # CloudFront requires SHA-1 hash
cf_signer = CloudFrontSigner(key_id, rsa_signer)
# To sign with a canned policy::
signed_url = cf_signer.generate_presigned_url(
url, date_less_than=datetime(2015, 12, 1))
# To sign with a custom policy::
signed_url = cf_signer.generate_presigned_url(url, policy=my_policy)
Disclaimer: I am the author of that PR.
As many have commented already, the initially accepted answer doesn't apply to Amazon CloudFront in fact, insofar Serving Private Content through CloudFront requires the use of dedicated CloudFront Signed URLs - accordingly secretmike's answer has been correct, but it is meanwhile outdated after he himself took the time and Added support for generating signed URLs for CloudFront (thanks much for this!).
boto now supports a dedicated create_signed_url method and the former binary dependency M2Crypto has recently been replaced with a pure-Python RSA implementation as well, see Don't use M2Crypto for cloudfront URL signing.
As increasingly common, one can find one or more good usage examples within the related unit tests (see test_signed_urls.py), for example test_canned_policy(self) - see setUp(self) for the referenced variables self.pk_idand self.pk_str (obviously you'll need your own keys):
def test_canned_policy(self):
"""
Generate signed url from the Example Canned Policy in Amazon's
documentation.
"""
url = "http://d604721fxaaqy9.cloudfront.net/horizon.jpg?large=yes&license=yes"
expire_time = 1258237200
expected_url = "http://example.com/" # replaced for brevity
signed_url = self.dist.create_signed_url(
url, self.pk_id, expire_time, private_key_string=self.pk_str)
# self.assertEqual(expected_url, signed_url)
This is what I use for create a policy so that I can give access to multiple files with the same "signature":
import json
import rsa
import time
from base64 import b64encode
url = "http://your_domain/*"
expires = int(time.time() + 3600)
pem = """-----BEGIN RSA PRIVATE KEY-----
...
-----END RSA PRIVATE KEY-----"""
key_pair_id = 'ABX....'
policy = {}
policy['Statement'] = [{}]
policy['Statement'][0]['Resource'] = url
policy['Statement'][0]['Condition'] = {}
policy['Statement'][0]['Condition']['DateLessThan'] = {}
policy['Statement'][0]['Condition']['DateLessThan']['AWS:EpochTime'] = expires
policy = json.dumps(policy)
private_key = rsa.PrivateKey.load_pkcs1(pem)
signature = b64encode(rsa.sign(str(policy), private_key, 'SHA-1'))
print '?Policy=%s&Signature=%s&Key-Pair-Id=%s' % (b64encode(policy),
signature,
key_pair_id)
I can use it for all files under http://your_domain/* for example:
http://your_domain/image2.png?Policy...
http://your_domain/image2.png?Policy...
http://your_domain/file1.json?Policy...