s3storage.py

import base64
import hmac
import os
import time
from hashlib import sha256

from django.conf import settings

from storages.backends.s3boto import S3BotoStorage

# See https://github.com/jschneier/django-storages/issues/28

os.environ['S3_USE_SIGV4'] = 'True'


class S3Storage(S3BotoStorage):
    def __init__(self, *args, **kwargs):
        kwargs['host'] = 's3.%s.amazonaws.com' % settings.AWS_REGION
        super(S3Storage, self).__init__(*args, **kwargs)

    def form_post_params(self, key, expires_in=600, acl=None,
                         success_action_redirect=None,
                         storage_class='STANDARD'):
        date_full = time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())
        date_short = date_full[:8]
        credential = '%s/%s/%s/s3/aws4_request' % (
            settings.AWS_ACCESS_KEY_ID,
            date_short,
            settings.AWS_REGION,
        )
        fields = []

        # Generate policy document
        expiration = time.gmtime(int(time.time() + expires_in))
        conditions = [
            '{"bucket": "%s"}' % settings.AWS_STORAGE_BUCKET_NAME,
            '{"key": "%s"}' % key,
            '{"x-amz-algorithm": "AWS4-HMAC-SHA256"}',
            '{"x-amz-credential": "%s"}' % credential,
            '{"x-amz-date": "%s"}' % date_full,
        ]
        if acl:
            conditions.append('{"acl": "%s"}' % acl)
            fields.append({'name': 'acl', 'value': acl})
        if success_action_redirect:
            conditions.append(
                '{"success_action_redirect": "%s"}' % success_action_redirect)
            fields.append({
                'name': 'success_action_redirect',
                'value': success_action_redirect,
            })
        if storage_class:
            conditions.append('{"x-amz-storage-class": "%s"}' % storage_class)
            fields.append({
                'name': 'x-amz-storage-class',
                'value': storage_class,
            })
        policy = self.connection.build_post_policy(expiration, conditions)
        policy_b64 = base64.b64encode(policy)

        signature = self.signature(policy_b64, date_short)
        fields.append({'name': 'key', 'value': key})
        fields.append({'name': 'policy', 'value': policy_b64})
        fields.append({'name': 'x-amz-algorithm', 'value': 'AWS4-HMAC-SHA256'})
        fields.append({'name': 'x-amz-credential', 'value': credential})
        fields.append({'name': 'x-amz-date', 'value': date_full})
        fields.append({'name': 'x-amz-signature', 'value': signature})

        return {
            'action': 'https://%s/' % self.connection.calling_format.build_host(
                self.connection.server_name(),
                settings.AWS_STORAGE_BUCKET_NAME
            ),
            'fields': fields,
        }

    def signature(self, string_to_sign, date):
        key = settings.AWS_SECRET_ACCESS_KEY
        k_date = self._sign(('AWS4' + key).encode('utf-8'), date)
        k_region = self._sign(k_date, settings.AWS_REGION)
        k_service = self._sign(k_region, 's3')
        k_signing = self._sign(k_service, 'aws4_request')
        return self._sign(k_signing, string_to_sign, hex=True)

    def _sign(self, key, msg, hex=False):
        if not isinstance(key, bytes):
            key = key.encode('utf-8')
        sig = hmac.new(key, msg.encode('utf-8'), sha256)
        return sig.hexdigest() if hex else sig.digest()