diff --git a/google/cloud/storage/_signing.py b/google/cloud/storage/_signing.py index 0da075ddb..8151786e9 100644 --- a/google/cloud/storage/_signing.py +++ b/google/cloud/storage/_signing.py @@ -509,7 +509,7 @@ def generate_signed_url_v4( :type query_parameters: dict :param query_parameters: - (Optional) Additional query paramtersto be included as part of the + (Optional) Additional query parameters to be included as part of the signed URLs. See: https://cloud.google.com/storage/docs/xml-api/reference-headers#query @@ -585,8 +585,7 @@ def generate_signed_url_v4( if generation is not None: query_parameters["generation"] = generation - ordered_query_parameters = sorted(query_parameters.items()) - canonical_query_string = six.moves.urllib.parse.urlencode(ordered_query_parameters) + canonical_query_string = _url_encode(query_parameters) lowercased_headers = dict(ordered_headers) @@ -672,3 +671,34 @@ def _sign_message(message, access_token, service_account_email): data = json.loads(response.data.decode("utf-8")) return data["signature"] + + +def _url_encode(query_params): + """Encode query params into URL. + + :type query_params: dict + :param query_params: Query params to be encoded. + + :rtype: str + :returns: URL encoded query params. + """ + params = [ + "{}={}".format(_quote_param(name), _quote_param(value)) + for name, value in query_params.items() + ] + + return "&".join(sorted(params)) + + +def _quote_param(param): + """Quote query param. + + :type param: Any + :param param: Query param to be encoded. + + :rtype: str + :returns: URL encoded query param. + """ + if not isinstance(param, bytes): + param = str(param) + return six.moves.urllib.parse.quote(param, safe="~") diff --git a/tests/unit/test__signing.py b/tests/unit/test__signing.py index c7231b56b..47ed2bdf0 100644 --- a/tests/unit/test__signing.py +++ b/tests/unit/test__signing.py @@ -1,3 +1,5 @@ +# -*- coding: utf-8 -*- +# # Copyright 2017 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -705,6 +707,61 @@ def test_sign_bytes_failure(self): ) +class TestCustomURLEncoding(unittest.TestCase): + def test_url_encode(self): + from google.cloud.storage._signing import _url_encode + + # param1 includes safe symbol ~ + # param# includes symbols, which must be encoded + query_params = {"param1": "value~1-2", "param#": "*value+value/"} + + self.assertEqual( + _url_encode(query_params), "param%23=%2Avalue%2Bvalue%2F¶m1=value~1-2" + ) + + +class TestQuoteParam(unittest.TestCase): + def test_ascii_symbols(self): + from google.cloud.storage._signing import _quote_param + + encoded_param = _quote_param("param") + self.assertIsInstance(encoded_param, str) + self.assertEqual(encoded_param, "param") + + def test_quoted_symbols(self): + from google.cloud.storage._signing import _quote_param + + encoded_param = _quote_param("!#$%&'()*+,/:;=?@[]") + self.assertIsInstance(encoded_param, str) + self.assertEqual( + encoded_param, "%21%23%24%25%26%27%28%29%2A%2B%2C%2F%3A%3B%3D%3F%40%5B%5D" + ) + + def test_unquoted_symbols(self): + from google.cloud.storage._signing import _quote_param + import string + + UNQUOTED = string.ascii_letters + string.digits + ".~_-" + + encoded_param = _quote_param(UNQUOTED) + self.assertIsInstance(encoded_param, str) + self.assertEqual(encoded_param, UNQUOTED) + + def test_unicode_symbols(self): + from google.cloud.storage._signing import _quote_param + + encoded_param = _quote_param("ЁЙЦЯЩЯЩ") + self.assertIsInstance(encoded_param, str) + self.assertEqual(encoded_param, "%D0%81%D0%99%D0%A6%D0%AF%D0%A9%D0%AF%D0%A9") + + def test_bytes(self): + from google.cloud.storage._signing import _quote_param + + encoded_param = _quote_param(b"bytes") + self.assertIsInstance(encoded_param, str) + self.assertEqual(encoded_param, "bytes") + + _DUMMY_SERVICE_ACCOUNT = None @@ -731,6 +788,7 @@ def _run_conformance_test(resource, test_data): method=test_data["method"], _request_timestamp=test_data["timestamp"], headers=test_data.get("headers"), + query_parameters=test_data.get("queryParameters"), ) assert url == test_data["expectedUrl"]