CloudFront Signed Cookie generator in Python

README.md

CloudFront Signed Cookie generator

For signed URLs, refer here

Usage

from cloudfront_signed_url import generate_cloudfront_signed_url

url = "https://your-cf-domain.com/path/to/file.txt"
cookie = generate_cloudfront_signed_cookie(url, 3600)
print(generate_curl_signed_cookies(url, cookie))

Signed Cookie generation is not implemented in boto3 (only Signed URLs). This gist attempts to make a minimal, simple, independent Signed Cookie generator for CloudFront (or a starting point for more complex logic).

You may use aws_base64_decode method to check generated policies.

Prerequisites

• Configured CloudFront Distribution
• An Origin access identity and a CloudFront key
• Origin and Behavior configured to Restrict Viewer Access

Reference: http://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/PrivateContent.html

Parameters

• Make sure you have a CloudFront key pair and key group created and associated with distribution behavior, details here)
• Replace newlines with \\n in private key and save the result to an SSM Parameter Store paramater named CF_SIGNED_URL_PRIVATE_KEY
• Save key ID (key pair ID) to a parameter named CF_SIGNED_URL_KEY_ID

CloudFront Signed Cookies

Signed Cookie consists of:

• CloudFront-Policy: Access policy, encoded with base64 (RFC-restricted characters replaced)
• CloudFront-Signature: Access policy, encrypted with a private key and encoded with base64 (RFC-restricted characters replaced)
• CloudFront-Key-Pair-Id: CloudFront key-pair ID

Access policy must be stripped to not contain any whitespace, and must be a valid JSON.

Notes

• Unlike S3 pre-signed URLs, you can use cookies generated once multiple times, as long as it is still valid (TTL).
• You can modify make_policy to make other policies (not per-url, but broader clauses), or even signed cookies (Policy, Signature, Key-Pair-Id are generated in the same way for cookies too).

Dependencies

• cryptography
• boto3

cloudfront-signed-cookie-generator.py

import base64
import datetime
import json

import boto3
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding

ssm_client = boto3.client('ssm')

# CloudFront secret from SSM
CF_SIGNED_URL_KEY_PAIR_ID = ssm_client.get_parameter(f"CF_SIGNED_URL_KEY_PAIR_ID")
CF_SIGNED_URL_PRIVATE_KEY = ssm_client.get_parameter(f"CF_SIGNED_URL_PRIVATE_KEY")

# fix possible escaped newlines
CF_SIGNED_URL_PRIVATE_KEY = CF_SIGNED_URL_PRIVATE_KEY.replace(b'\\n', b'\n')

# or, define your own


def make_cloudfront_signature(message, secret):
    private_key = serialization.load_pem_private_key(
        secret,
        password=None,
        backend=default_backend()
    )
    return private_key.sign(message, padding.PKCS1v15(), hashes.SHA1())


def make_cloudfront_policy(resource, expire_date):
    policy = {
        'Statement': [{
            'Resource': resource,
            'Condition': {
                'DateLessThan': {
                    'AWS:EpochTime': expire_date
                }
            }
        }]
    }
    return json.dumps(policy).replace(" ", "")


def aws_base64_encode(data):
    return base64.b64encode(data).replace(b'+', b'-').replace(b'=', b'_').replace(b'/', b'~')


def aws_base64_decode(data):
    return base64.b64decode(data.replace(b'-', b'+').replace(b'_', b'=').replace(b'~', b'/'))


def generate_cloudfront_signature(policy, private_key):
    policy = policy.encode('utf8')
    signature = make_cloudfront_signature(policy, private_key)
    return signature


def generate_cloudfront_signed_cookie(resource: str, expire_seconds: int):
    expiration = (datetime.datetime.now() + datetime.timedelta(seconds=expire_seconds)).timestamp()
    expiration = int(expiration)
    policy = make_cloudfront_policy(resource, expiration)
    signature = generate_cloudfront_signature(policy, CF_SIGNED_URL_PRIVATE_KEY)

    return {
        'CloudFront-Policy': aws_base64_encode(policy),
        'CloudFront-Signature': aws_base64_encode(signature),
        'CloudFront-Key-Pair-Id': CF_SIGNED_URL_KEY_PAIR_ID
    }

Completely broken. Not in a working state. The call on line 70 to the author's own function isn't even valid: it's missing key_id.

@mrcrilly Hi, thanks for pointing this out. I rewrote all logic with types and tested on real distributions, seems working. What do you think?

@prog893 thanks for the gist. I found one error. aws_base64_encode expects bytes but aws_base64_encode(policy) uses policy:str

def generate_cloudfront_signature(policy, private_key):
    policy = policy.encode('utf8')
    signature = make_cloudfront_signature(policy, private_key)
    return signature


def generate_cloudfront_signed_cookie(resource: str, expire_seconds: int):
    expiration = (datetime.datetime.now() + datetime.timedelta(seconds=expire_seconds)).timestamp()
    expiration = int(expiration)
    policy = make_cloudfront_policy(resource, expiration)
    signature = generate_cloudfront_signature(policy, CF_SIGNED_URL_PRIVATE_KEY)

    return {
        'CloudFront-Policy': aws_base64_encode(policy.encode('utf8')),
        'CloudFront-Signature': aws_base64_encode(signature),
        'CloudFront-Key-Pair-Id': CF_SIGNED_URL_KEY_PAIR_ID
    }

I figured out how to do this without all the encoding. You can build a custom policy using botocore's CloudFrontSigner

# Encode the private key as bytes
CF_PRIVATE_KEY = CF_PRIVATE_KEY.encode("utf-8")

# fix possible escaped newlines
CF_PRIVATE_KEY = CF_PRIVATE_KEY.replace(b"\\n", b"\n")

EXPIRATION_SECONDS = 600
CLOUDFRONT_BASE_URL = "https://mycf.cloudfront.net"


def get_cf_sign(message):
    private_key = serialization.load_pem_private_key(
        CF_PRIVATE_KEY, password=None, backend=default_backend()
    )
    return private_key.sign(message, padding.PKCS1v15(), hashes.SHA1())


cloudfront_signer = CloudFrontSigner(CF_KEY_PAIR_ID, get_cf_sign)


def fetch_signed_url():
    file_path = Video.objects.first().file_path
    mpd_url = f"{CLOUDFRONT_BASE_URL}/{file_path}"
    video_url = f"{CLOUDFRONT_BASE_URL}/{file_path.replace('.mpd', '_video.mp4')}"
    audio_url = f"{CLOUDFRONT_BASE_URL}/{file_path.replace('.mpd', '_audio.mp4')}"

    # Replace ".mpd" with "*"
    full_url = mpd_url.replace(".mpd", "*")
    expires_at = datetime.datetime.now() + datetime.timedelta(
        seconds=EXPIRATION_SECONDS
    )

    policy = cloudfront_signer.build_policy(full_url, expires_at)
    cf_signed_url = cloudfront_signer.generate_presigned_url(full_url, policy=policy)

    query_string = cf_signed_url.split("?")[1]
    query_params = dict(qc.split("=") for qc in query_string.split("&"))

    cookie = {
        "CloudFront-Policy": query_params["Policy"],
        "CloudFront-Signature": query_params["Signature"],
        "CloudFront-Key-Pair-Id": query_params["Key-Pair-Id"],
    }

    response = requests.get(audio_url, cookies=cookie, stream=True)

    if response.ok:
        return cookie
    else:
        print(response.reason)
        return None