Fix: Generate Signed Urls through a service account by providing service_account_email and access_token

docs/backends/gcloud.rst

@@ -53,7 +53,16 @@ In most cases, the default service accounts are not sufficient to read/write and
 #. Make sure your service account has access to the bucket and appropriate permissions. (`Using IAM Permissions <https://cloud.google.com/storage/docs/access-control/using-iam-permissions>`__)
 #. Ensure this service account is associated to the type of compute being used (Google Compute Engine (GCE), Google Kubernetes Engine (GKE), Google Cloud Run (GCR), etc)
 
-For development use cases, or other instances outside Google infrastructure:
+**Note:** There is currently a limitation in the GCS client for Python which by default requires a service account private key file to be
+present when generating signed urls. The service account private key is unavailable when running on a compute service.
+Compute Services (App Engine, Cloud Run, Cloud Functions, Compute Engine...) fetch `access tokens from the metadata server <https://cloud.google.com/docs/authentication/application-default-credentials>`__ .
+These services do not have access to the service account private key. This means that when trying to sign data in these services,
+you **MUST** use Cloud IAM sign function (SignBlob) to sign data and directly signing data isn't possible by any means.
+
+Luckily this can be worked around by passing `service_account_email` and `access_token` to the generate_signed_url function.
+When both of those args are provided, generate_signed_url will use the IAM SignBlob API to sign the url and no private key file is needed.
+
+Last resort you can still use the service account key file for authentication (not recommended by Google):
 
 #. Create the key and download ``your-project-XXXXX.json`` file.
 #. Ensure the key is mounted/available to your running Django app.
@@ -219,3 +228,21 @@ Settings
   It supports `timedelta`, `datetime`, or `integer` seconds since epoch time.
 
   Note: The maximum value for this option is 7 days (604800 seconds) in version `v4` (See this `Github issue  <https://github.com/googleapis/python-storage/issues/456#issuecomment-856884993>`_)
+
+``iam_sign_blob`` or ``GS_IAM_SIGN_BLOB``
+
+  default: ``False``
+
+  Signing urls requires a service account key file to be present in the env or IAM SignBlob API call
+  through a service account email and access_token. Certain GCP services (ex: Compute services) don't have access to the key file in the env.
+  This setting needs to be `True` when running on such services as they fetch access tokens from metadata server instead of having key files
+  If using `v4` of generate_signed_url, `google-cloud-storage>=v1.36.1 <https://github.com/googleapis/python-storage/releases/tag/v1.36.1>`_ is required .
+
+``sa_email`` or ``GS_SA_EMAIL``
+
+  default: ``None``
+
+  The service account email to use for signing url. If a service account is being used for authentication (attached to your service),
+  this setting doesn't need to be provided unless you want to use another service account than the one attached to your service for signing urls.
+  Can be used in local development env as well to sign using sa_email instead of the user credentials or keeping a insecure service account key file
+  If using `v4` of generate_signed_url, `google-cloud-storage>=v1.36.1 <https://github.com/googleapis/python-storage/releases/tag/v1.36.1>`_ is required .

pyproject.toml

@@ -54,7 +54,7 @@ dropbox = [
   "dropbox>=7.2.1",
 ]
 google = [
-  "google-cloud-storage>=1.27",
+  "google-cloud-storage>=1.36.1",
 ]
 libcloud = [
   "apache-libcloud",

storages/backends/gcloud.py

@@ -19,6 +19,9 @@
 from storages.utils import to_bytes
 
 try:
+    from google import auth
+    from google.auth.transport import requests
+    from google.auth.credentials import TokenState
     from google.cloud.exceptions import NotFound
     from google.cloud.storage import Blob
     from google.cloud.storage import Client
@@ -141,12 +144,22 @@ def get_default_settings(self):
             # roll over.
             "max_memory_size": setting("GS_MAX_MEMORY_SIZE", 0),
             "blob_chunk_size": setting("GS_BLOB_CHUNK_SIZE"),
+            # use in cases where service account key isn't available in env
+            # in such cases, sign blob api is REQUIRED for signing data
+            "iam_sign_blob": setting("GS_IAM_SIGN_BLOB", False),
+            "sa_email": setting("GS_SA_EMAIL"),
         }
 
     @property
     def client(self):
         if self._client is None:
+            if self.project_id is None or self.credentials is None:
+                self.credentials, self.project_id = auth.default(
+                    scopes=['https://www.googleapis.com/auth/cloud-platform']
+                )
             self._client = Client(project=self.project_id, credentials=self.credentials)
+        if self.credentials and self.credentials.token_state != TokenState.FRESH:
+            self.credentials.refresh(requests.Request())
         return self._client
 
     @property
@@ -330,6 +343,19 @@ def url(self, name, parameters=None):
             }
             params = parameters or {}
 
+            if self.iam_sign_blob:
+                if not hasattr(self.credentials, "service_account_email") and not self.sa_email:
+                    raise AttributeError(
+                        "Sign Blob API requires service_account_email to be available "
+                        "through ADC or setting `sa_email`"
+                    )
+                if hasattr(self.credentials, "service_account_email"):
+                    default_params["service_account_email"] = self.credentials.service_account_email
+                # sa_email has the final say of which service_account_email to be used for signing if provided
+                if self.sa_email:
+                    default_params["service_account_email"] = self.sa_email
+                default_params["access_token"] = self.credentials.token
+
             for key, value in default_params.items():
                 if value and key not in params:
                     params[key] = value

tests/test_gcloud.py

@@ -36,6 +36,7 @@ def test_open_read(self):
         """
         data = b"This is some test read data."
 
+        self.storage._client = mock.MagicMock()
         with self.storage.open(self.filename) as f:
             self.storage._client.bucket.assert_called_with(self.bucket_name)
             self.storage._bucket.get_blob.assert_called_with(
@@ -49,6 +50,7 @@ def test_open_read_num_bytes(self):
         data = b"This is some test read data."
         num_bytes = 10
 
+        self.storage._client = mock.MagicMock()
         with self.storage.open(self.filename) as f:
             self.storage._client.bucket.assert_called_with(self.bucket_name)
             self.storage._bucket.get_blob.assert_called_with(
@@ -107,6 +109,7 @@ def test_save(self):
         data = "This is some test content."
         content = ContentFile(data)
 
+        self.storage._client = mock.MagicMock()
         self.storage.save(self.filename, content)
 
         self.storage._client.bucket.assert_called_with(self.bucket_name)
@@ -124,6 +127,7 @@ def test_save2(self):
         filename = "ủⓝï℅ⅆℇ.txt"
         content = ContentFile(data)
 
+        self.storage._client = mock.MagicMock()
         self.storage.save(filename, content)
 
         self.storage._client.bucket.assert_called_with(self.bucket_name)
@@ -145,7 +149,7 @@ def test_save_with_default_acl(self):
         # 'projectPrivate', 'bucketOwnerRead', 'bucketOwnerFullControl',
         # 'private', 'authenticatedRead', 'publicRead', 'publicReadWrite'
         self.storage.default_acl = "publicRead"
-
+        self.storage._client = mock.MagicMock()
         self.storage.save(filename, content)
 
         self.storage._client.bucket.assert_called_with(self.bucket_name)
@@ -159,6 +163,7 @@ def test_save_with_default_acl(self):
         )
 
     def test_delete(self):
+        self.storage._client = mock.MagicMock()
         self.storage.delete(self.filename)
 
         self.storage._client.bucket.assert_called_with(self.bucket_name)
@@ -185,6 +190,7 @@ def test_exists_no_bucket(self):
 
     def test_exists_bucket(self):
         # exists('') should return True if the bucket exists
+        self.storage._client = mock.MagicMock()
         self.assertTrue(self.storage.exists(""))
 
     def test_exists_file_overwrite(self):
@@ -371,6 +377,7 @@ def test_custom_endpoint_with_parameters(self):
 
         self.storage.default_acl = "publicRead"
         url = "{}/{}".format(self.storage.custom_endpoint, self.filename)
+        self.storage._client = mock.MagicMock()
         self.assertEqual(self.storage.url(self.filename), url)
 
         bucket_name = "hyacinth"
@@ -414,6 +421,7 @@ def test_cache_control(self):
             GS_OBJECT_PARAMETERS={"cache_control": "public, max-age=604800"}
         ):
             self.storage = gcloud.GoogleCloudStorage(bucket_name=self.bucket_name)
+            self.storage._client = mock.MagicMock()
             self.storage.save(filename, content)
             bucket = self.storage.client.bucket(self.bucket_name)
             blob = bucket.get_blob(filename)
@@ -427,6 +435,7 @@ def test_storage_save_gzip_twice(self):
         content = ContentFile("I should be gzip'd")
 
         # When
+        self.storage._client = mock.MagicMock()
         self.storage.save(name, content)
         self.storage.save("test_storage_save_2.css", content)
 
@@ -528,6 +537,7 @@ def test_storage_save_gzipped(self, *args):
         patcher = mock.patch("google.cloud.storage.Bucket.get_blob", return_value=blob)
         try:
             patcher.start()
+            self.storage._client = mock.MagicMock()
             self.storage.save(name, content)
             obj = self.storage._bucket.get_blob()
             obj.upload_from_file.assert_called_with(
@@ -555,6 +565,7 @@ def test_storage_save_gzip(self, *args):
 
         try:
             patcher.start()
+            self.storage._client = mock.MagicMock()
             self.storage.save(name, content)
             obj = self.storage._bucket.get_blob()
             obj.upload_from_file.assert_called_with(