Repository: googleapis/python-storage

Issue:
Impossible to use signed url v4 and the Service Account Credential APIs
Months ago, I solved an issue to a Stack overflow user by explaining [how to generate a signed URL without a service account JSON fey file](https://stackoverflow.com/questions/64234214/how-to-generate-a-blob-signed-url-in-google-cloud-run/64245028#64245028). I also [release an article on this](https://medium.com/google-cloud/service-account-credentials-api-a-solution-to-different-issues-dc7434037115)

Today, a comment of a user told me that is doesn't work. They open a new [Stack Overflow question](https://stackoverflow.com/questions/65568916/generating-cloud-storage-signed-url-from-google-cloud-function-without-using-exp/65569603#65569603) and in their question, there is the v4 mentioned.

I dug into the code and yes, the v2 and the v4 haven't the same behavior: the v4 check if the credential can [generate a signed URL by itself](https://github.com/googleapis/python-storage/blob/9d5bb5622f17a0143fa6670406596e53d0068286/google/cloud/storage/_signing.py#L536), that implies the [useless check later](https://github.com/googleapis/python-storage/blob/9d5bb5622f17a0143fa6670406596e53d0068286/google/cloud/storage/_signing.py#L623) that allows to use Service Account Credential API, instead of the service account private key to generate the signature.


#### Environment details

  - Linux (Cloud Function and Cloud Run)
  - Python version: `python --version` 3.7 and 3.8
  - `google-cloud-storage` version: latest (1.35.0)

#### Steps to reproduce

  1. Working code on Cloud Run (for example) *Replace YOUR_BUCKET by a valid bucket*

```
import os
from flask import Flask, request

app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def sign_url():
    from google.cloud import storage
    from datetime import datetime, timedelta

    import google.auth

    credentials, project_id = google.auth.default()

    # Perform a refresh request to get the access token of the current credentials (Else, it's None)
    from google.auth.transport import requests
    r = requests.Request()
    credentials.refresh(r)

    client = storage.Client()
    bucket = client.get_bucket('YOUR_BUCKET')
    blob = bucket.get_blob('name.csv')

    service_account_email = credentials.service_account_email

    url = blob.generate_signed_url(service_account_email=service_account_email, access_token=credentials.token,version="v2", method="PUT",expiration=timedelta(minutes=120),content_type="application/octet-stream")
    return url, 200

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

```

  2.  Broken code on Cloud Run (for example) *Replace YOUR_BUCKET by a valid bucket*

```
import os
from flask import Flask, request

app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def sign_url():
    from google.cloud import storage
    from datetime import datetime, timedelta

    import google.auth

    credentials, project_id = google.auth.default()

    # Perform a refresh request to get the access token of the current credentials (Else, it's None)
    from google.auth.transport import requests
    r = requests.Request()
    credentials.refresh(r)

    client = storage.Client()
    bucket = client.get_bucket('YOUR_BUCKET')
    blob = bucket.get_blob('name.csv')

    service_account_email = credentials.service_account_email

    url = blob.generate_signed_url(service_account_email=service_account_email, access_token=credentials.token,version="v4", method="PUT",expiration=timedelta(minutes=120),content_type="application/octet-stream")
    return url, 200

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

```

*Only the version (v2/v4) has changed*

Fix proposal is coming.

Discussion:
cajoek: I really want this!
guillaumeblaquiere: Sorry for the delay, the week was awful!
guillaumeblaquiere: @tseaver any review possible on this?

Patch:
diff --git a/google/cloud/storage/_signing.py b/google/cloud/storage/_signing.py
index f2e94ba..16c9397 100644
--- a/google/cloud/storage/_signing.py
+++ b/google/cloud/storage/_signing.py
@@ -457,9 +457,12 @@ def generate_signed_url_v4(
                    google-cloud-python/issues/922
     .. _reference: https://cloud.google.com/storage/docs/reference-headers
 
+
     :type credentials: :class:`google.auth.credentials.Signing`
     :param credentials: Credentials object with an associated private key to
-                        sign text.
+                        sign text. That credentials must provide signer_email
+                        only if service_account_email and access_token are not
+                        passed.
 
     :type resource: str
     :param resource: A pointer to a specific resource
@@ -533,7 +536,6 @@ def generate_signed_url_v4(
     :returns: A signed URL you can use to access the resource
               until expiration.
     """
-    ensure_signed_credentials(credentials)
     expiration_seconds = get_expiration_seconds_v4(expiration)
 
     if _request_timestamp is None:
@@ -542,7 +544,11 @@ def generate_signed_url_v4(
         request_timestamp = _request_timestamp
         datestamp = _request_timestamp[:8]
 
-    client_email = credentials.signer_email
+    client_email = service_account_email
+    if not access_token or not service_account_email:
+        ensure_signed_credentials(credentials)
+        client_email = credentials.signer_email
+
     credential_scope = "{}/auto/storage/goog4_request".format(datestamp)
     credential = "{}/{}".format(client_email, credential_scope)
 


Test patch:
diff --git a/tests/unit/test__signing.py b/tests/unit/test__signing.py
index 7cf817c..3eac70c 100644
--- a/tests/unit/test__signing.py
+++ b/tests/unit/test__signing.py
@@ -653,7 +653,22 @@ class Test_generate_signed_url_v4(unittest.TestCase):
     def test_w_custom_query_parameters_w_none_value(self):
         self._generate_helper(query_parameters={"qux": None})
 
-    def test_with_access_token(self):
+    def test_with_access_token_and_service_account_email(self):
+        resource = "/name/path"
+        credentials = _make_credentials()
+        email = mock.sentinel.service_account_email
+        with mock.patch(
+            "google.cloud.storage._signing._sign_message", return_value=b"DEADBEEF"
+        ):
+            self._call_fut(
+                credentials,
+                resource=resource,
+                expiration=datetime.timedelta(days=5),
+                service_account_email=email,
+                access_token="token",
+            )
+
+    def test_with_access_token_and_service_account_email_and_signer_email(self):
         resource = "/name/path"
         signer_email = "service@example.com"
         credentials = _make_credentials(signer_email=signer_email)
@@ -668,6 +683,39 @@ class Test_generate_signed_url_v4(unittest.TestCase):
                 access_token="token",
             )
 
+    def test_with_signer_email(self):
+        resource = "/name/path"
+        signer_email = "service@example.com"
+        credentials = _make_credentials(signer_email=signer_email)
+        credentials.sign_bytes.return_value = b"DEADBEEF"
+        self._call_fut(
+            credentials, resource=resource, expiration=datetime.timedelta(days=5),
+        )
+
+    def test_with_service_account_email_and_signer_email(self):
+        resource = "/name/path"
+        signer_email = "service@example.com"
+        credentials = _make_credentials(signer_email=signer_email)
+        credentials.sign_bytes.return_value = b"DEADBEEF"
+        self._call_fut(
+            credentials,
+            resource=resource,
+            expiration=datetime.timedelta(days=5),
+            service_account_email=signer_email,
+        )
+
+    def test_with_token_and_signer_email(self):
+        resource = "/name/path"
+        signer_email = "service@example.com"
+        credentials = _make_credentials(signer_email=signer_email)
+        credentials.sign_bytes.return_value = b"DEADBEEF"
+        self._call_fut(
+            credentials,
+            resource=resource,
+            expiration=datetime.timedelta(days=5),
+            access_token="token",
+        )
+
 
 class Test_sign_message(unittest.TestCase):
     @staticmethod