Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: allow signed url version v4 without signed credentials #356

Merged
merged 9 commits into from Feb 19, 2021
12 changes: 9 additions & 3 deletions google/cloud/storage/_signing.py
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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
guillaumeblaquiere marked this conversation as resolved.
Show resolved Hide resolved

credential_scope = "{}/auto/storage/goog4_request".format(datestamp)
credential = "{}/{}".format(client_email, credential_scope)

Expand Down
50 changes: 49 additions & 1 deletion tests/unit/test__signing.py
Expand Up @@ -653,7 +653,22 @@ def test_w_custom_query_parameters_w_string_value(self):
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)
Expand All @@ -668,6 +683,39 @@ def test_with_access_token(self):
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
Expand Down