Add dummy backend to run s3file in development

Author: codingjoe

README.rst

@@ -106,8 +106,8 @@ Progress Bar
 ------------
 
 S3File does emit progress signals that can be used to display some kind of progress bar.
-Signals named `progress` are emitted for both each individual file input as well as for
-the form as a whole.
+Signals named ``progress`` are emitted for both each individual file input as well as
+for the form as a whole.
 
 The progress signal carries the following details:
 
@@ -151,6 +151,24 @@ entire form.
         })
     })()
 
+
+Using S3File in development
+---------------------------
+
+Using S3File in development can be helpful especially if you want to use the progress
+signals described above. Therefore, S3File comes with a AWS S3 dummy backend.
+It behaves similar to the real S3 storage backend. It is automatically enabled, if the
+``DEFAULT_FILE_STORAGE`` setting is set to ``FileSystemStorage``.
+
+To prevent users from accidentally using the ``FileSystemStorage`` and the insecure S3
+dummy backend in production, there is also an additional deployment check that will
+error if you run Django's deployment check suite::
+
+    python manage.py check --deploy
+
+We recommend always running the deployment check suite as part of your deployment
+pipeline.
+
 Uploading multiple files
 ------------------------
 

s3file/apps.py

@@ -1,4 +1,7 @@
 from django.apps import AppConfig
+from django.core import checks
+
+from .checks import storage_check
 
 
 class S3FileConfig(AppConfig):
@@ -9,6 +12,7 @@ def ready(self):
         from django.core.files.storage import default_storage
         from storages.backends.s3boto3 import S3Boto3Storage
         from django import forms
+
         from .forms import S3FileInputMixin
 
         if isinstance(default_storage, S3Boto3Storage) and \
@@ -21,3 +25,5 @@ def ready(self):
                 cls for cls in forms.ClearableFileInput.__bases__
                 if cls is not S3FileInputMixin
             )
+
+        checks.register(storage_check, checks.Tags.security, deploy=True)

s3file/checks.py

@@ -0,0 +1,13 @@
+from django.core.checks import Error
+from django.core.files.storage import FileSystemStorage, default_storage
+
+
+def storage_check(app_configs, **kwargs):
+    if isinstance(default_storage, FileSystemStorage):
+        return [
+            Error(
+                'FileSystemStorage should not be used in a production environment.',
+                hint='Please verify your DEFAULT_FILE_STORAGE setting.',
+                id='s3file.E001',
+            )
+        ]

s3file/forms.py

@@ -4,9 +4,10 @@
 import uuid
 
 from django.conf import settings
-from django.core.files.storage import default_storage
 from django.utils.functional import cached_property
 
+from s3file.storages import storage
+
 logger = logging.getLogger('s3file')
 
 
@@ -21,11 +22,11 @@ class S3FileInputMixin:
 
     @property
     def bucket_name(self):
-        return default_storage.bucket.name
+        return storage.bucket.name
 
     @property
     def client(self):
-        return default_storage.connection.meta.client
+        return storage.connection.meta.client
 
     def build_attrs(self, *args, **kwargs):
         attrs = super().build_attrs(*args, **kwargs)

s3file/middleware.py

@@ -1,7 +1,9 @@
 import logging
 import os
 
-from django.core.files.storage import default_storage
+from s3file.storages import local_dev, storage
+
+from . import views
 
 logger = logging.getLogger('s3file')
 
@@ -17,14 +19,17 @@ def __call__(self, request):
             paths = request.POST.getlist(field_name, [])
             request.FILES.setlist(field_name, list(self.get_files_from_storage(paths)))
 
+        if local_dev and request.path == '/__s3_mock__/':
+            return views.S3MockView.as_view()(request)
+
         return self.get_response(request)
 
     @staticmethod
     def get_files_from_storage(paths):
         """Return S3 file where the name does not include the path."""
         for path in paths:
             try:
-                f = default_storage.open(path)
+                f = storage.open(path)
                 f.name = os.path.basename(path)
                 yield f
             except OSError:

s3file/storages.py

@@ -0,0 +1,50 @@
+import base64
+import datetime
+import hmac
+import json
+
+from django.conf import settings
+from django.core.files.storage import FileSystemStorage, default_storage
+
+
+class S3MockStorage(FileSystemStorage):
+    class connection:
+        class meta:
+            class client:
+                @staticmethod
+                def generate_presigned_post(bucket_name, key, **policy):
+                    policy = json.dumps(policy).encode()
+                    policy_b64 = base64.b64encode(policy).decode()
+                    date = datetime.datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')
+                    aws_id = getattr(
+                        settings, 'AWS_ACCESS_KEY_ID',
+                        'AWS_ACCESS_KEY_ID',
+                    )
+                    fields = {
+                        'x-amz-algorithm': 'AWS4-HMAC-SHA256',
+                        'x-amz-date': date,
+                        'x-amz-credential': aws_id,
+                        'policy': policy_b64,
+                        'key': key,
+                    }
+                    signature = hmac.new(
+                        settings.SECRET_KEY.encode(),
+                        policy + date.encode(),
+                        'sha256',
+                    ).digest()
+                    signature = base64.b64encode(signature).decode()
+                    return {
+                        'url': '/__s3_mock__/',
+                        'fields': {
+                            'x-amz-signature': signature,
+                            **fields
+                        },
+                    }
+
+    class bucket:
+        name = 'test-bucket'
+
+
+local_dev = isinstance(default_storage, FileSystemStorage)
+
+storage = default_storage if not local_dev else S3MockStorage()

s3file/views.py

@@ -0,0 +1,58 @@
+import base64
+import hashlib
+import hmac
+import logging
+
+from django import http
+from django.conf import settings
+from django.core.files.storage import default_storage
+from django.views import generic
+
+logger = logging.getLogger('s3file')
+
+
+class S3MockView(generic.View):
+
+    def post(self, request):
+        success_action_status = request.POST.get('success_action_status', 201)
+        try:
+            file = request.FILES['file']
+            key = request.POST['key']
+            date = request.POST['x-amz-date']
+            signature = request.POST['x-amz-signature']
+            policy = request.POST['policy']
+        except KeyError:
+            logger.exception('bad request')
+            return http.HttpResponseBadRequest()
+
+        try:
+            signature = base64.b64decode(signature.encode())
+            policy = base64.b64decode(policy.encode())
+
+            calc_sign = hmac.new(
+                settings.SECRET_KEY.encode(),
+                policy + date.encode(),
+                'sha256'
+            ).digest()
+        except ValueError:
+            logger.exception('bad request')
+            return http.HttpResponseBadRequest()
+
+        if not hmac.compare_digest(signature, calc_sign):
+            logger.warning('bad signature')
+            return http.HttpResponseForbidden()
+
+        key = key.replace('${filename}', file.name)
+        etag = hashlib.md5(file.read()).hexdigest()  # nosec
+        file.seek(0)
+        key = default_storage.save(key, file)
+        return http.HttpResponse(
+            "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+            "<PostResponse>"
+            f"<Location>{settings.MEDIA_URL}{key}</Location>"
+            f"<Bucket>{getattr(settings, 'AWS_STORAGE_BUCKET_NAME')}</Bucket>"
+            f"<Key>{key}</Key>"
+            f"<ETag>\"{etag}\"</ETag>"
+            "</PostResponse>",
+            status=success_action_status,
+        )

tests/test_checks.py

@@ -0,0 +1,14 @@
+import pytest
+from django.core.management import call_command
+from django.core.management.base import SystemCheckError
+
+
+def test_storage_check():
+    call_command('check')
+
+    with pytest.raises(SystemCheckError) as e:
+        call_command('check', '--deploy')
+
+    assert (
+        'FileSystemStorage should not be used in a production environment.'
+    ) in str(e.value)

tests/test_views.py

@@ -0,0 +1,69 @@
+import base64
+import hmac
+import http
+
+from s3file import views
+
+
+class TestS3MockView:
+    url = '/__s3_mock__/'
+
+    policy = (
+        'eyJDb25kaXRpb25zIjogW3siYnVja2V0IjogInRlc3QtYnVja2V0In0sIFsic3RhcnRz'
+        'LXdpdGgiLCAiJGtleSIsICJ0bXAvczNmaWxlLzNlUWhwOTZYU1dldFFwZ1VVQmZzWHci'
+        'XSwgeyJzdWNjZXNzX2FjdGlvbl9zdGF0dXMiOiAiMjAxIn0sIFsic3RhcnRzLXdpdGgi'
+        'LCAiJENvbnRlbnQtVHlwZSIsICIiXV0sICJFeHBpcmVzSW4iOiAxMjA5NjAwfQ=='
+    )
+    date = '20190616T184210Z'
+
+    def test_post__bad_request(self, rf):
+        request = rf.post(self.url, data={})
+        response = views.S3MockView.as_view()(request)
+        assert response.status_code == http.HTTPStatus.BAD_REQUEST
+
+    def test_post__created(self, client, upload_file):
+        with open(upload_file) as fp:
+            response = client.post(self.url, data={
+                'x-amz-signature': 'T1Mb71D45h1u2SBoUHOMBNFCI2AgrKAg1UzKdUHUHig=',
+                'x-amz-algorithm': 'AWS4-HMAC-SHA256',
+                'x-amz-date': self.date,
+                'x-amz-credential': 'testaccessid',
+                'policy': self.policy,
+                'key': 'tmp/s3file/3eQhp96XSWetQpgUUBfsXw/${filename}',
+                'success_action_status': '201',
+                'file': fp,
+            })
+        assert response.status_code == http.HTTPStatus.CREATED
+
+    def test_post__bad_signature(self, client, upload_file):
+
+        bad_signature = base64.b64encode(
+            hmac.new(b'eve', (self.policy + self.date).encode(), 'sha256').digest()
+        ).decode()
+        with open(upload_file) as fp:
+            response = client.post(self.url, data={
+                'x-amz-signature': bad_signature,
+                'x-amz-algorithm': 'AWS4-HMAC-SHA256',
+                'x-amz-date': self.date,
+                'x-amz-credential': 'testaccessid',
+                'policy': self.policy,
+                'key': 'tmp/s3file/3eQhp96XSWetQpgUUBfsXw/${filename}',
+                'success_action_status': '201',
+                'file': fp,
+            })
+        assert response.status_code == http.HTTPStatus.FORBIDDEN
+
+    def test_post__not_a_signature(self, client, upload_file):
+        bad_signature = 'eve'
+        with open(upload_file) as fp:
+            response = client.post(self.url, data={
+                'x-amz-signature': bad_signature,
+                'x-amz-algorithm': 'AWS4-HMAC-SHA256',
+                'x-amz-date': self.date,
+                'x-amz-credential': 'testaccessid',
+                'policy': self.policy,
+                'key': 'tmp/s3file/3eQhp96XSWetQpgUUBfsXw/${filename}',
+                'success_action_status': '201',
+                'file': fp,
+            })
+        assert response.status_code == http.HTTPStatus.BAD_REQUEST