Cloudfront Signed Cookie

[keyute]

While signed cloudfront url solves the issue of serving private files from S3 via Cloudfront, it is very slow when a user wants to access many resources at once such as a landing page. Is there a way to use signed cookie instead?

[0xRaduan]

Hey, @keyute, do you mind sharing any snippets on how you've solved this problem for yourself?

[tony]

@keyute @Raduan77 If you or anyone reading this ever comes across a way of doing CloudFront Cookie signing for files on django, it'd be great to know how you set it up!

Especially when it comes to handling it with localhost development / other domains / etc.

[krystofbe]

Greetings everyone,

I hope you find the following information helpful, especially for those who may come across this issue in the future. Although it has been nearly three years since the original post, I wanted to share the cloudfront django middleware that we have been utilizing to enable authenticated users to access our cloudfront CDN:

class CloudFrontMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.cloudfront_util = CloudFrontUtil(settings.AWS_CLOUDFRONT_KEY, settings.AWS_CLOUDFRONT_KEY_ID)

    def __call__(self, request):
        response = self.get_response(request)

        # only set the cookie for authenticated users
        if request.user.is_authenticated:
            # Example cookie names returned by `generate_signed_cookies` method
            cookie_names = ["CloudFront-Policy", "CloudFront-Signature", "CloudFront-Key-Pair-Id"]
            # If any of the cookies are not present in the request, generate them
            if not all(cookie in request.COOKIES for cookie in cookie_names):
                # Expire from settings session expire time
                expire_at = datetime.utcnow() + timedelta(seconds=settings.SESSION_COOKIE_AGE)
                # You could provide specific URL or wildcard depending on your requirements
                url = "*"
                signed_cookies = self.cloudfront_util.generate_signed_cookies(url, expire_at)

                for cookie_name, cookie_value in signed_cookies.items():
                    response.set_cookie(
                        cookie_name, cookie_value, domain=settings.AWS_CLOUDFRONT_COOKIE_DOMAIN, httponly=True
                    )

        return response


class CloudFrontUtil:
    def __init__(self, private_key: str, key_id: str):
        """
        :param private_key: str, the string of private key which generated by openssl command line
        :param key_id: str, CloudFront -> Key management -> Public keys
        """
        self.key_id = key_id

        priv_key = rsa.PrivateKey.load_pkcs1(private_key)

        # NOTE: CloudFront use RSA-SHA1 for signing URLs or cookies
        self.rsa_signer = functools.partial(rsa.sign, priv_key=priv_key, hash_method="SHA-1")
        self.cf_signer = CloudFrontSigner(key_id, self.rsa_signer)

    def generate_signed_cookies(self, url: str, expire_at: datetime) -> str:
        policy = self.cf_signer.build_policy(url, expire_at).encode("utf8")
        policy_64 = self.cf_signer._url_b64encode(policy).decode("utf8")

        signature = self.rsa_signer(policy)
        signature_64 = self.cf_signer._url_b64encode(signature).decode("utf8")
        return {
            "CloudFront-Policy": policy_64,
            "CloudFront-Signature": signature_64,
            "CloudFront-Key-Pair-Id": self.key_id,
        }

[ckkz-it]

Thanks @krystofbe for the code, it helped me a lot. One thing though, the rsa library didn't work for my PEM private key (which I generated following AWS docs), so I had to use the cryptography library instead. In case somebody interested in the working code for me:

import base64
from datetime import datetime, timedelta

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from botocore.signers import CloudFrontSigner
from django.conf import settings
from django.core.handlers.wsgi import WSGIRequest
from django.utils import timezone


class CloudFrontMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.cloudfront_util = CloudFrontUtil(
            settings.AWS_CLOUDFRONT_KEY, settings.AWS_CLOUDFRONT_KEY_ID
        )

    def __call__(self, request: WSGIRequest):
        response = self.get_response(request)

        if not request.user.is_authenticated:
            return response

        cookie_names = [
            "CloudFront-Policy",
            "CloudFront-Signature",
            "CloudFront-Key-Pair-Id",
        ]
        if all(cookie_name in request.COOKIES for cookie_name in cookie_names):
            return response

        expire_at = timezone.now() + timedelta(days=1)
        url = f"https://{settings.AWS_CLOUDFRONT_URL}/user_{request.user.id}/*"
        signed_cookies = self.cloudfront_util.generate_signed_cookies(url, expire_at)

        for cookie_name, cookie_value in signed_cookies.items():
            response.set_cookie(
                cookie_name,
                cookie_value,
                domain=settings.AWS_CLOUDFRONT_URL,
                httponly=True,
                secure=True,
            )

        return response


class CloudFrontUtil:
    def __init__(self, private_key: bytes, key_id: str):
        """
        :param private_key: bytes, the bytes string of private key which is generated by openssl command line
        :param key_id: str, CloudFront -> Key management -> Public keys
        """
        self.key_id = key_id

        key: RSAPrivateKey = load_pem_private_key(
            private_key, password=None, backend=default_backend()
        )

        self.rsa_signer = lambda x: key.sign(x, padding.PKCS1v15(), hashes.SHA1())
        self.cf_signer = CloudFrontSigner(key_id, self.rsa_signer)

    def generate_signed_cookies(self, url: str, expire_at: datetime) -> dict[str, str]:
        policy = self.cf_signer.build_policy(url, expire_at).encode("utf8")
        policy_64 = self._url_b64encode(policy)

        signature = self.rsa_signer(policy)
        signature_64 = self._url_b64encode(signature)
        return {
            "CloudFront-Policy": policy_64,
            "CloudFront-Signature": signature_64,
            "CloudFront-Key-Pair-Id": self.key_id,
        }

    def _url_b64encode(self, data: bytes) -> str:
        # Required by CloudFront. See also:
        # http://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/private-content-linux-openssl.html
        return (
            base64.b64encode(data)
            .replace(b"+", b"-")
            .replace(b"=", b"_")
            .replace(b"/", b"~")
        ).decode("utf8")

I took the signer part from django storages lib

django-storages/storages/backends/s3.py

Line 46 in 70fa8d9

def _use_cryptography_signer():