S3PrivateFileField

__init__.py

#

README.md

Description

Utilities for private/protected file storage.

This field is a dropin replacement for FileField, but the files require you to be logged in.

S3PrivateFileField

Add to urls:

url(r'^private-files/', include('alpaca.private_storage.urls')),

Use field:

def check_file_permission(user, obj):
    if user.is_authenticated and obj.user == user
    
class MyModel(Model):
    my_file = S3PrivateFileField(permission_func=check_file_permission)

Reference the file url like normal:

{{ my_object.my_file.url }}

If you need the original S3 url for some reason, access it with _private_url:

{{ my_object.my_file._private_url }}

However, if you do that, that is kind of missing the point of the private file.

Note: for this to work, the S3 bucket needs to be using Access Control List (ACL) and NOT a Bucket Policy. Usually, you can just delete any bucket policy there, and the ACL will be used.

fields.py

from base64 import urlsafe_b64encode

from django.db.models.fields.files import FieldFile, FileField
from django.urls import reverse

from storages.backends.s3boto import S3BotoStorage


private_storage = S3BotoStorage(
    acl='private',
    querystring_auth=True,
    querystring_expire=60,
)


class S3PrivateFieldFile(FieldFile):
    def _get_url(self):
        instance = self.instance
        app_label = instance._meta.app_label
        model_name = instance._meta.model_name
        field_name = self.field.name
        combined_string = '{}.{}:{}:{}'.format(app_label, model_name, instance.pk, field_name)
        combined_string = urlsafe_b64encode(bytes(combined_string, 'utf8'))

        return reverse(
            'private_storage:private_file_redirect',
            kwargs={'key': combined_string, 'file_name': self.name}
        )
    url = property(_get_url)

    def _get_private_url(self):
        return super(S3PrivateFieldFile, self)._get_url()
    _private_url = property(_get_private_url)


class S3PrivateFileField(FileField):
    attr_class = S3PrivateFieldFile
    permission_func = None

    def __init__(self, verbose_name=None, name=None, upload_to='', storage=None, permission_func=None, **kwargs):

        if permission_func:
            self.permission_func = permission_func
        else:
            self.permission_func = lambda user, obj: user.is_authenticated

        super(S3PrivateFileField, self).__init__(
            verbose_name=verbose_name,
            name=name,
            upload_to=upload_to,
            storage=storage or private_storage,
            **kwargs
        )


try:
    from model_mommy import mommy
    mommy.generators.add(S3PrivateFileField, mommy.random_gen.gen_file_field)
except ImportError:
    # if model mommy is not installed, don't worry about registering the field
    pass

urls.py

"""
Urls for private_storage app
"""

from django.conf.urls import url

from . import views

app_name = 'private_storage'
urlpatterns = [
    url(r'^(?P<key>[=\w]+)/(?P<file_name>.+)', views.PrivateFileRedirect.as_view(), name='private_file_redirect'),
]

views.py

from base64 import urlsafe_b64decode

from django.apps import apps
from django.contrib.auth.mixins import AccessMixin
from django.shortcuts import get_object_or_404, redirect
from django.http.response import HttpResponseForbidden
from django.views import View


class PrivateFileRedirect(AccessMixin, View):
    def get(self, request, *args, **kwargs):
        key = kwargs['key']
        combined_string = urlsafe_b64decode(bytes(key, 'utf8')).decode('utf-8')

        app_model, instance_pk, field_name = combined_string.split(':')
        app_label, model_name = app_model.split('.')

        model = apps.get_model(app_label, model_name)
        field = model._meta.get_field(field_name)
        instance = get_object_or_404(model, pk=instance_pk)

        permission_func = field.permission_func
        is_allowed = permission_func(self.request.user, instance)

        if is_allowed:
            return redirect(getattr(instance, field_name)._private_url)
        else:
            return self.handle_no_permission()

    def handle_no_permission(self):
        if self.request.user.is_authenticated:
            return HttpResponseForbidden()

        return super(PrivateFileRedirect, self).handle_no_permission()