Python3 Utility to create Signed URLs for Google Cloud Storage File Upload (with expiry)

gcp_signed_url_util.py

"""
Code to create a signed URL to upload a file using that signed URL to Google Cloud Storage
Constants to be present in settings module or can be defined locally:
- GCP_CRED_FILE_PATH (JSON file where the GCP credentials are present)
- GCP_PROJECT (GCP Project name)
- GCP_STORAGE_BASE_URL (Base Url of GCP, usually https://storage.googleapis.com)
- GCP_SIGNED_URL_TIMEOUT (Timeout in seconds for which the signed URL is valid)
- GCP_BUCKET (Google Cloud Bucket Name)
"""
import logging
import time
import base64
from google.oauth2 import service_account
from google.cloud import datastore
# Gettings settings variables
from django.conf import settings

# Google Account Credentials & Datastore Client
GOOGLE_SERVICE_ACCOUNT_CREDS = service_account.Credentials.from_service_account_file(settings.GCP_CRED_FILE_PATH)
GOOGLE_DATASTORE_CLIENT = datastore.Client(project=settings.GCP_PROJECT,
                                           credentials=GOOGLE_SERVICE_ACCOUNT_CREDS)


def make_string_to_sign(verb, expiration, resource,
                        content_md5="", content_type="", content_headers=""):
    """
    Function to create the string to sign, from the parameters given.
    Refer: https://cloud.google.com/storage/docs/access-control/signed-urls
    Args:
        verb: HTTP verb to be used with the signed URL
        expiration: timestamp reprsented as epoch of time when signature expires.
                    Refer: https://en.wikipedia.org/wiki/Unix_time
        resource: Resource being addressed in URL, in format /bucket/objname.
                  Ex. /kartik-tesst/Macbook-3D.jpg
        content_md5: MD5 digest value in base64. if used, client must provide this in header
                     when making the request to the generated signed URL
        content_type: content-type to be accessed. if used, client must provide this in header
                     when making the request to the generated signed URL
        content_headers: When client uses signed URL, server will check for matching values as
                         those provided in this field
    Returns:
        string_to_sign: Returns string that is to be signed

    """

    string_to_sign = (
        '{verb}\n'
        '{content_md5}\n'
        '{content_type}\n'
        '{expiration}\n'
        '{headers}'
        '{resource}'
        ).format(verb=verb, content_md5=content_md5,
                 content_type=content_type, expiration=expiration,
                 headers=content_headers, resource=resource)
    # Logging
    logging.debug("String to Sign = {sts}".format(sts=string_to_sign))
        
    return string_to_sign

def sign_string(string_to_sign, gcp_credentials=GOOGLE_SERVICE_ACCOUNT_CREDS):
    """
    Function to sign a given string, using the specified keyfile
    Refer: https://cloud.google.com/storage/docs/access-control/create-signed-urls-program
    Args:
        string_to_sign: The string that is to be signed, created according to the
                        required syntax.
        keyfile_path: Path to keyfile of Google Service Account.
                      Refer: https://cloud.google.com/storage/docs/authentication#service_accounts
    Returns:
        (client_id, signed_string)
        client_id: service account email that was used to sign the string
        signed_string: base64 encoded string after signing
    """
    #TODO: Add error handling when credentials do not allow access to specified resource in string
    #TODO: Add error handling when credentials are invalid
    # Logging
    logging.debug("Signing string!")
    creds = gcp_credentials
    client_id = creds.service_account_email
    signed_string = creds.sign_bytes(string_to_sign.encode('utf-8'))
    signed_string = str(base64.b64encode(signed_string))
    # Logging
    logging.debug("Signed String = {ss}".format(ss=str(signed_string[2:-1])))
    return (client_id, str(signed_string[2:-1]))

def convert_url_safe(in_str):
    """
    Converts the input string to a URL safe string
    Args:
        in_str: Input string to be made URL safe
    Returns:
        out_str: Output URL safe string
    """
    temp = in_str.replace('+', '%2B')
    out_str = temp.replace('/', '%2F')
    # Logging
    logging.debug("URL Safe String = {uss}".format(uss=out_str))
    return out_str

def make_url(base, access_id, expiration, signature):
    """
    Creates the signed URL from its parts
    Args:
        base: settings.GCP_STORAGE_BASE_URL of the signed URL
        access_id: GOOGLE_ACCESS_ID for storage
        expiration: Expiration time in Epoch Unix Time
        signature: signed string
        Refer: https://cloud.google.com/storage/docs/access-control/create-signed-urls-program
    Returns:
        signed_url: Signed URL
    """
    # Logging
    logging.debug("Making the URL!")
    signed_url = "{base}?GoogleAccessId={access_id}&Expires={exp_time}&Signature={sign}"
    signed_url = signed_url.format(base=base, access_id=access_id,
                                   exp_time=expiration, sign=signature)
    # Logging
    logging.debug("Signed URL = {su}".format(su=signed_url))
    return signed_url

def generate_signed_url(uuid, content_headers="", content_md5="",
                        content_type="", gcp_credentials=GOOGLE_SERVICE_ACCOUNT_CREDS,
                        valid_time=settings.GCP_SIGNED_URL_TIMEOUT):
    """
    Function to generate the signed URL to upload a file to Google Cloud Storage.
    Args:
        uuid: The unique uuid or filename of the file to be uploaded to Google CLoud Storage.
        bucket: The Google Cloud Storage bucket to which the file is to be uploaded.
                The files are always at the first level,ie file will be uploaded at <bucket>/<file>.
        content_headers: When client uses signed URL, server will check for matching values as 
                         those provided in this field
        content_md5: MD5 digest value in base64. if used, client must provide this in header
                     when making the request to the generated signed URL
        content_type: The content-type of the file to be uploaded. If this is specified,
                      then content-type must be specified in the header when uploading the
                      file using the signed URL.
        gcp_credentials: The Google Service Account Credentials that are used to sign the URL.
        valid_time: Time( in seconds) for which the signed URL is valid
    Returns:
        signed_url: The signed URL as a string.
    """
    # Logging
    logging.debug("Starting signed URL generation!")
    c_time = int(time.time())
    e_time = c_time + valid_time
    test_string = make_string_to_sign(verb="PUT", expiration=str(e_time),
                                      resource="/{bucket}/"\
                                               "{_file}".format(bucket=settings.GCP_BUCKET,
                                                                _file=uuid),
                                      content_type=content_type, content_md5=content_md5,
                                      content_headers=content_headers)
    try:
        client_id, signed_string = sign_string(test_string, gcp_credentials)
    except Exception:
        # Logging
        logging.error("There was an error while signing the URL!")
        raise ValueError("There was an error while signing the URL!")

    sign_safe = convert_url_safe(signed_string)
    base = settings.GCP_STORAGE_BASE_URL + "/{buck}/{_file}".format(buck=settings.GCP_BUCKET,
                                                                    _file=uuid)
    url = make_url(base=base, access_id=client_id, expiration=e_time, signature=sign_safe)
    # Logging
    logging.debug("Signed URL Generated!")
    return url