Skip to content

Instantly share code, notes, and snippets.

@ipmb
Created March 25, 2020 17:09
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save ipmb/3007bc8e977bad858f38772731d6e4dc to your computer and use it in GitHub Desktop.
Save ipmb/3007bc8e977bad858f38772731d6e4dc to your computer and use it in GitHub Desktop.
Django S3 Storage backend to generate signed Cloudfront URLs
"""
S3 Storage served via private Cloudfront URL
https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/PrivateContent.html
https://github.com/boto/boto3/blob/develop/boto3/examples/cloudfront.rst
"""
import datetime
from botocore.signers import CloudFrontSigner
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from django.conf import settings
from django.utils import timezone as tz
from django.utils.deconstruct import deconstructible
from django.utils.encoding import filepath_to_uri
from storages.backends.s3boto3 import S3Boto3Storage
def _rsa_signer(message: str) -> bytearray:
pk = serialization.load_pem_private_key(
settings.AWS_CLOUDFRONT_PRIVATE_KEY.strip().encode("utf-8"),
password=None,
backend=default_backend(),
)
return pk.sign(message, padding.PKCS1v15(), hashes.SHA1())
@deconstructible
class CloudfrontS3Boto3Storage(S3Boto3Storage):
def url(self, name, parameters=None, expire=None):
# Preserve the trailing slash after normalizing the path.
name = self._normalize_name(self._clean_name(name))
if self.custom_domain:
return "%s//%s/%s" % (
self.url_protocol,
self.custom_domain,
filepath_to_uri(name),
)
if expire is None:
expire = self.querystring_expire
cloudfront_signer = CloudFrontSigner(
settings.AWS_CLOUDFRONT_KEY_ID, _rsa_signer
)
url = cloudfront_signer.generate_presigned_url(
"%s//%s/%s"
% (
self.url_protocol,
settings.AWS_CLOUDFRONT_DOMAIN,
filepath_to_uri(name),
),
date_less_than=tz.now() + datetime.timedelta(seconds=expire),
)
if self.querystring_auth:
return url
return self._strip_signing_parameters(url)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment