fix: make v4 signing formatting consistent w/ spec

google/cloud/storage/_signing.py

@@ -18,7 +18,6 @@
 import collections
 import datetime
 import hashlib
-import re
 import json
 
 import six
@@ -31,8 +30,6 @@
 
 
 NOW = datetime.datetime.utcnow  # To be replaced by tests.
-MULTIPLE_SPACES_RE = r"\s+"
-MULTIPLE_SPACES = re.compile(MULTIPLE_SPACES_RE)
 
 SERVICE_ACCOUNT_URL = (
     "https://googleapis.dev/python/google-api-core/latest/"
@@ -192,7 +189,7 @@ def get_canonical_headers(headers):
     normalized = collections.defaultdict(list)
     for key, val in headers:
         key = key.lower().strip()
-        val = MULTIPLE_SPACES.sub(" ", val.strip())
+        val = " ".join(val.split())
         normalized[key].append(val)
 
     ordered_headers = sorted((key, ",".join(val)) for key, val in normalized.items())
@@ -206,8 +203,8 @@ def get_canonical_headers(headers):
 )
 
 
-def canonicalize(method, resource, query_parameters, headers):
-    """Canonicalize method, resource
+def canonicalize_v2(method, resource, query_parameters, headers):
+    """Canonicalize method, resource per the V2 spec.
 
     :type method: str
     :param method: The HTTP verb that will be used when requesting the URL.
@@ -368,7 +365,7 @@ def generate_signed_url_v2(
     """
     expiration_stamp = get_expiration_seconds_v2(expiration)
 
-    canonical = canonicalize(method, resource, query_parameters, headers)
+    canonical = canonicalize_v2(method, resource, query_parameters, headers)
 
     # Generate the string to sign.
     elements_to_sign = [
@@ -589,13 +586,22 @@ def generate_signed_url_v4(
     ordered_query_parameters = sorted(query_parameters.items())
     canonical_query_string = six.moves.urllib.parse.urlencode(ordered_query_parameters)
 
+    quoted_resource = six.moves.urllib.parse.quote(resource)
+
+    lowercased_headers = dict(ordered_headers)
+
+    if "x-goog-content-sha256" in lowercased_headers:
+        payload = lowercased_headers["x-goog-content-sha256"]
+    else:
+        payload = "UNSIGNED-PAYLOAD"
+
     canonical_elements = [
         method,
-        resource,
+        quoted_resource,
         canonical_query_string,
         canonical_header_string,
         signed_headers,
-        "UNSIGNED-PAYLOAD",
+        payload,
     ]
     canonical_request = "\n".join(canonical_elements)
 
@@ -620,7 +626,7 @@ def generate_signed_url_v4(
         signature = binascii.hexlify(signature_bytes).decode("ascii")
 
     return "{}{}?{}&X-Goog-Signature={}".format(
-        api_access_endpoint, resource, canonical_query_string, signature
+        api_access_endpoint, quoted_resource, canonical_query_string, signature
     )
 
 

tests/unit/test__signing.py

@@ -309,12 +309,12 @@ def test_w_embedded_ws(self):
         self.assertEqual(ordered, expected_ordered)
 
 
-class Test_canonicalize(unittest.TestCase):
+class Test_canonicalize_v2(unittest.TestCase):
     @staticmethod
     def _call_fut(*args, **kwargs):
-        from google.cloud.storage._signing import canonicalize
+        from google.cloud.storage._signing import canonicalize_v2
 
-        return canonicalize(*args, **kwargs)
+        return canonicalize_v2(*args, **kwargs)
 
     def test_wo_headers_or_query_parameters(self):
         method = "GET"
@@ -544,9 +544,9 @@ def _generate_helper(
         generation=None,
         headers=None,
         query_parameters=None,
+        resource="/name/path",
     ):
         now = datetime.datetime(2019, 2, 26, 19, 53, 27)
-        resource = "/name/path"
         signer_email = "service@example.com"
         credentials = _make_credentials(signer_email=signer_email)
         credentials.sign_bytes.return_value = b"DEADBEEF"
@@ -577,7 +577,7 @@ def _generate_helper(
         )
         self.assertEqual(scheme, expected_scheme)
         self.assertEqual(netloc, expected_netloc)
-        self.assertEqual(path, resource)
+        self.assertEqual(path, six.moves.urllib.parse.quote(resource))
         self.assertEqual(frag, "")
 
         # Check the URL parameters.
@@ -650,12 +650,18 @@ def test_w_custom_host_header(self):
     def test_w_custom_headers(self):
         self._generate_helper(headers={"x-goog-foo": "bar"})
 
+    def test_w_custom_payload_hash_goog(self):
+        self._generate_helper(headers={"x-goog-content-sha256": "DEADBEEF"})
+
     def test_w_custom_query_parameters_w_string_value(self):
         self._generate_helper(query_parameters={"bar": "/"})
 
     def test_w_custom_query_parameters_w_none_value(self):
         self._generate_helper(query_parameters={"qux": None})
 
+    def test_w_non_ascii_resource(self):
+        self._generate_helper(resource="/name/p\xe5th")
+
     def test_with_access_token(self):
         resource = "/name/path"
         signer_email = "service@example.com"