storage.py

"""This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license.
"""
import logging
import os
import threading
from urllib.parse import unquote, urlsplit, urlunsplit

import google.auth
from django.conf import settings
from django.contrib.staticfiles.storage import ManifestStaticFilesStorage
from storages.backends.azure_storage import AzureStorage
from storages.backends.gcloud import GoogleCloudStorage, _quote, clean_name
from storages.backends.s3boto3 import S3Boto3Storage

logger = logging.getLogger(__name__)


class SkipMissedManifestStaticFilesStorage(ManifestStaticFilesStorage):
    """We need this class to escape missing files from
    django.contrib.staticfiles.finders.FileSystemFinder:
    this class tries to find js/css/png/jpg/... inside of you js/css/...
    """

    # Disable strict cache manifest checking
    manifest_strict = False

    def hashed_name(self, name, content=None, filename=None):
        # `filename` is the name of file to hash if `content` isn't given.
        # `name` is the base name to construct the new hashed filename from.
        parsed_name = urlsplit(unquote(name))
        clean_name = parsed_name.path.strip()
        filename = (filename and urlsplit(unquote(filename)).path.strip()) or clean_name
        opened = content is None
        if opened:
            if not self.exists(filename):
                return ''
            try:
                content = self.open(filename)
            except IOError:
                # Handle directory paths and fragments
                return name
        try:
            file_hash = self.file_hash(clean_name, content)
        finally:
            if opened:
                content.close()
        path, filename = os.path.split(clean_name)
        root, ext = os.path.splitext(filename)
        if file_hash is not None:
            file_hash = '.%s' % file_hash
        hashed_name = os.path.join(path, '%s%s%s' % (root, file_hash, ext))
        unparsed_name = list(parsed_name)
        unparsed_name[2] = hashed_name
        # Special casing for a @font-face hack, like url(myfont.eot?#iefix")
        # http://www.fontspring.com/blog/the-new-bulletproof-font-face-syntax
        if '?#' in name and not unparsed_name[3]:
            unparsed_name[2] += '?'
        return urlunsplit(unparsed_name)


class StorageProxyMixin:
    def url(self, name, storage_url=False, *args, **kwargs):
        if storage_url is True:
            return super().url(name, *args, **kwargs)
        return f'{settings.HOSTNAME}/storage-data/uploaded/?filepath={name}'


class CustomS3Boto3Storage(StorageProxyMixin, S3Boto3Storage):
    def url(self, name, storage_url=False, *args, **kwargs):
        if storage_url is True:
            return super().url(name, *args, **kwargs)
        return f"s3://{settings.AWS_STORAGE_BUCKET_NAME}/{name}"


class CustomAzureStorage(StorageProxyMixin, AzureStorage):
    pass


class AlternativeGoogleCloudStorageBase(GoogleCloudStorage):
    """A subclass to force the use of the IAM signBlob API
    This allows the signing of blob URLs without having to use a credential file.
    The service account must have the iam.serviceAccounts.signBlob permission."""

    def __init__(self, **settings):
        super().__init__(**settings)
        self._signing_credentials = None
        self._signing_credentials_lock = threading.Lock()

    def url(self, name):
        """
        Return public url or a signed url for the Blob.
        This DOES NOT check for existence of Blob - that makes codes too slow
        for many use cases.
        Overridden to force the use of the IAM signBlob API.
        See https://github.com/googleapis/python-storage/blob/519074112775c19742522158f612b467cf590219/google/cloud/storage/_signing.py#L628  # NOQA
        """
        name = self._normalize_name(clean_name(name))
        blob = self.bucket.blob(name)
        blob_params = self.get_object_parameters(name)
        no_signed_url = blob_params.get('acl', self.default_acl) == 'publicRead' or not self.querystring_auth

        if not self.custom_endpoint and no_signed_url:
            return blob.public_url
        elif no_signed_url:
            out = '{storage_base_url}/{quoted_name}'.format(
                storage_base_url=self.custom_endpoint,
                quoted_name=_quote(name, safe=b'/~'),
            )
            return out
        elif not self.custom_endpoint:
            out2 = blob.generate_signed_url(expiration=self.expiration, version='v4', **self._get_signing_kwargs())
            return out2
        else:
            out3 = blob.generate_signed_url(
                bucket_bound_hostname=self.custom_endpoint,
                expiration=self.expiration,
                version='v4',
                **self._get_signing_kwargs(),
            )
            return out3

    def _get_signing_credentials(self):
        with self._signing_credentials_lock:
            if self._signing_credentials is None or self._signing_credentials.expired:
                credentials, _ = google.auth.default(['https://www.googleapis.com/auth/cloud-platform'])
                auth_req = google.auth.transport.requests.Request()
                credentials.refresh(auth_req)
                self._signing_credentials = credentials
        return self._signing_credentials

    def _get_signing_kwargs(self):
        credentials = self._get_signing_credentials()
        out = {
            'service_account_email': credentials.service_account_email,
            'access_token': credentials.token,
            'credentials': credentials,
        }
        return out


class AlternativeGoogleCloudStorage(StorageProxyMixin, AlternativeGoogleCloudStorageBase):
    pass