Utilities for upload and download files to Google Cloud Storage using signed URLs within Google App Engine applications

cloud_storage.py

import datetime
import time
import urllib

__all__ = ['sign', 'PolicyDocument', 'CloudStorageURLSigner']

from google.appengine.api import app_identity
from base64 import b64encode
import json


def sign(string_to_sign):
    signing_key_name, signature = app_identity.sign_blob(string_to_sign)
    return b64encode(signature)


class PolicyDocument:
    """Represents a policy.

    Attributes:
        content_type:
        success_action_redirect:
        key:
        bucket:
        expiration:
        acl:
        success_action_status:
    """
    ACL = "acl"
    SUCCESS_ACTION_REDIRECT = "success_action_redirect"
    SUCCESS_ACTION_STATUS = "success_action_status"
    KEY = "key"
    BUCKET = "bucket"
    CONTENT_TYPE = "content-type"
    ACL_PUBLIC_READ = "public-read"
    ACL_PROJECT_PRIVATE = "project-private"

    def __init__(self, content_type=None, success_action_redirect=None, key=None, bucket=None, expiration=None,
                 success_action_status=201, acl=ACL_PROJECT_PRIVATE):
        self.content_type = content_type
        self.success_action_redirect = success_action_redirect
        self.key = key
        self.bucket = bucket
        self.expiration = expiration
        self.acl = acl
        self.success_action_status = success_action_status

    def as_dict(self):
        conditions = [{self.ACL: self.acl},
                      {self.BUCKET: self.bucket},
                      {self.KEY: self.key},
                      {self.CONTENT_TYPE: self.content_type},
                      ["starts-with", "$content-type", 'image/'],
        ]

        if self.success_action_redirect:
            conditions.append({self.SUCCESS_ACTION_REDIRECT: self.success_action_redirect})
        else:
            conditions.append({self.SUCCESS_ACTION_STATUS: str(self.success_action_status)})

        return dict(expiration=self.expiration, conditions=conditions)

    def as_json_b64encode(self):
        return b64encode(self.as_json())

    def as_json(self):
        return json.dumps(self.as_dict())


class CloudStorageURLSigner(object):
    """Contains methods for generating signed URLs for Google Cloud Storage."""

    DEFAULT_GCS_API_ENDPOINT = 'https://storage.googleapis.com'

    def __init__(self, gcs_api_endpoint=None, expiration=None):
        """Creates a CloudStorageURLSigner that can be used to access signed URLs.
    Args:
      gcs_api_endpoint: Base URL for GCS API. Default is 'https://storage.googleapis.com'
      expiration: An instance of datetime.datetime containing the time when the
                  signed URL should expire.
    """
        self.gcs_api_endpoint = gcs_api_endpoint or self.DEFAULT_GCS_API_ENDPOINT
        self.expiration = expiration or (datetime.datetime.now() +
                                         datetime.timedelta(days=1))
        self.expiration = int(time.mktime(self.expiration.timetuple()))
        self.client_id_email = app_identity.get_service_account_name()

    def __make_signature_string(self, verb, path, content_md5, content_type):
        """Creates the signature string for signing according to GCS docs."""
        signature_string = ('{verb}\n'
                            '{content_md5}\n'
                            '{content_type}\n'
                            '{expiration}\n'
                            '{resource}')
        return signature_string.format(verb=verb,
                                       content_md5=content_md5,
                                       content_type=content_type,
                                       expiration=self.expiration,
                                       resource=path)

    def signed_url(self, verb, path, content_type='', content_md5=''):
        """Forms and returns the full signed URL to access GCS."""
        base_url = '%s%s' % (self.gcs_api_endpoint, path)
        signature_string = self.__make_signature_string(verb, path, content_md5,
                                                        content_type)
        signature = urllib.quote_plus(sign(signature_string))
        return "{}?GoogleAccessId={}&Expires={}&Signature={}".format(base_url, self.client_id_email,
                                                                     str(self.expiration), signature)