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

docs/backends/gcloud.rst

@@ -52,15 +52,40 @@ In most cases, the default service accounts are not sufficient to read/write and
 #. Create a service account. (`Google Getting Started Guide <https://cloud.google.com/docs/authentication/getting-started>`__)
 #. 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)
+#. If your django app only handles ``publicRead`` storage objects then, above steps are all that is required
+#. If your django app handles signed (expiring) urls, then read through the options in the ``Settings for Signed Urls`` section
 
-For development use cases, or other instances outside Google infrastructure:
+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.
 #. Set an environment variable of GOOGLE_APPLICATION_CREDENTIALS to the path of the json file.
 
 Alternatively, you can use the setting ``credentials`` or ``GS_CREDENTIALS`` as described below.
 
+Settings for Signed Urls
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. 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 file is unavailable when running on compute services. 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>`__
+
+Due to the above limitation, currently the only way to generate signed url without having the private key file mounted
+in the env is through the IAM Sign Blob API.
+
+IAM Sign Blob API doesn't require a private key file to be present in the env, but it does have
+`quota limits <https://cloud.google.com/iam/quotas#quotas>`__ which could be a deal-breaker. In order to enable this,
+the setting ``GS_IAM_SIGN_BLOB`` (default=`False`) needs to be `True`. When this setting is enabled,
+signed urls are generated through the IAM SignBlob API using the attached service account email and access_token instead
+of the credentials in the key file.
+
+``GS_IAM_SIGN_BLOB`` setting is also complemented with the optional setting ``GS_SA_EMAIL``. This setting allows
+you to override the service account to be used to generate the signed url if it is different from the one attached
+to your env. Also useful for local/development use cases where the metadata server isn't available and storing private key
+files is dangerous.
 
 Settings
 ~~~~~~~~
@@ -219,3 +244,18 @@ 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``
+
+  Generate signed urls using the IAM Sign Blob API which doesn't require a service account private key file to be present in the env.
+  Set this setting to ``True`` if storing private key file isn't viable and would rather generate signed urls using an API.
+
+``sa_email`` or ``GS_SA_EMAIL``
+
+  default: ``None``
+
+  Allows override of the service account to be used for generating signed urls using the IAM Sign Blob API.
+  This setting is completely optional and should be used if the service account associated with your service/app isn't
+  the one with the permissions to SignBlob. Also helpful for development use cases where private key file is not recommended.

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.credentials import TokenState
+    from google.auth.transport import requests
     from google.cloud.exceptions import NotFound
     from google.cloud.storage import Blob
     from google.cloud.storage import Client
@@ -141,11 +144,19 @@ 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.iam_sign_blob and not self.credentials:
+                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)
         return self._client
 
@@ -330,8 +341,34 @@ def url(self, name, parameters=None):
             }
             params = parameters or {}
 
+            if self.iam_sign_blob:
+                service_account_email, access_token = self._get_iam_sign_blob_params()
+                default_params["service_account_email"] = service_account_email
+                default_params["access_token"] = access_token
+
             for key, value in default_params.items():
                 if value and key not in params:
                     params[key] = value
 
             return blob.generate_signed_url(**params)
+
+    def _get_iam_sign_blob_params(self):
+        if self.credentials.token_state != TokenState.FRESH:
+            self.credentials.refresh(requests.Request())
+
+        try:
+            service_account_email = self.credentials.service_account_email
+        except AttributeError:
+            service_account_email = None
+
+        # sa_email has final say of service_account used to sign url if provided
+        if self.sa_email:
+            service_account_email = self.sa_email
+
+        if not service_account_email:
+            raise AttributeError(
+                "Sign Blob API requires service_account_email to be available "
+                "through ADC or setting `sa_email`"
+            )
+
+        return service_account_email, self.credentials.token

tests/test_gcloud.py

@@ -509,6 +509,79 @@ def test_dupe_file_chunk_size(self):
                 self.filename, chunk_size=chunk_size
             )
 
+    def test_iam_sign_blob_setting(self):
+        self.assertEqual(self.storage.iam_sign_blob, False)
+        with override_settings(GS_IAM_SIGN_BLOB=True):
+            storage = gcloud.GoogleCloudStorage()
+            self.assertEqual(storage.iam_sign_blob, True)
+
+    def test_sa_email_setting(self):
+        self.assertEqual(self.storage.sa_email, None)
+        with override_settings(GS_SA_EMAIL="[email protected]"):
+            storage = gcloud.GoogleCloudStorage()
+            self.assertEqual(storage.sa_email, "[email protected]")
+
+    def test_iam_sign_blob_no_service_account_email_raises_attribute_error(self):
+        with override_settings(GS_IAM_SIGN_BLOB=True):
+            storage = gcloud.GoogleCloudStorage()
+            storage._bucket = mock.MagicMock()
+            storage.credentials = mock.MagicMock()
+            # deleting mocked attribute to simulate no service_account_email
+            del storage.credentials.service_account_email
+            # simulating access token
+            storage.credentials.token = "1234"
+            # no sa_email or adc service_account_email found
+            with self.assertRaises(
+                AttributeError,
+                msg=(
+                    "Sign Blob API requires service_account_email to be available "
+                    "through ADC or setting `sa_email`"
+                ),
+            ):
+                storage.url(self.filename)
+
+    def test_iam_sign_blob_with_adc_service_account_email(self):
+        with override_settings(GS_IAM_SIGN_BLOB=True):
+            storage = gcloud.GoogleCloudStorage()
+            storage._bucket = mock.MagicMock()
+            storage.credentials = mock.MagicMock()
+            # simulating adc service account email
+            storage.credentials.service_account_email = "[email protected]"
+            # simulating access token
+            storage.credentials.token = "1234"
+            blob = mock.MagicMock()
+            storage._bucket.blob.return_value = blob
+            storage.url(self.filename)
+            # called with adc service account email and access token
+            blob.generate_signed_url.assert_called_with(
+                expiration=timedelta(seconds=86400),
+                version="v4",
+                service_account_email=storage.credentials.service_account_email,
+                access_token=storage.credentials.token,
+            )
+
+    def test_iam_sign_blob_with_sa_email_setting(self):
+        with override_settings(
+            GS_IAM_SIGN_BLOB=True, GS_SA_EMAIL="[email protected]"
+        ):
+            storage = gcloud.GoogleCloudStorage()
+            storage._bucket = mock.MagicMock()
+            storage.credentials = mock.MagicMock()
+            # simulating adc service account email
+            storage.credentials.service_account_email = "[email protected]"
+            # simulating access token
+            storage.credentials.token = "1234"
+            blob = mock.MagicMock()
+            storage._bucket.blob.return_value = blob
+            storage.url(self.filename)
+            # called with sa_email as it has final say
+            blob.generate_signed_url.assert_called_with(
+                expiration=timedelta(seconds=86400),
+                version="v4",
+                service_account_email=storage.sa_email,
+                access_token=storage.credentials.token,
+            )
+
 
 class GoogleCloudGzipClientTests(GCloudTestCase):
     def setUp(self):