Created
March 25, 2020 17:09
-
-
Save ipmb/3007bc8e977bad858f38772731d6e4dc to your computer and use it in GitHub Desktop.
Django S3 Storage backend to generate signed Cloudfront URLs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
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