googleapis · crwilcox · Mar 7, 2020 · Feb 7, 2020 · Feb 10, 2020 · Feb 12, 2020
diff --git a/google/cloud/storage/_signing.py b/google/cloud/storage/_signing.py
@@ -510,7 +510,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
 
@@ -586,8 +586,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)
 
     canonical_elements = [
         method,
@@ -666,3 +665,33 @@ 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 = []
+    for name, value in query_params.items():
+        params.append(_quote_param(name) + "=" + _quote_param(value))
+
+    return "&".join(sorted(params))
+
+
+def _quote_param(param):
+    """Quote query param.
+
+    :type param: dict
+    :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
@@ -702,6 +702,19 @@ 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&param1=value~1-2"
+        )
+
+
 _DUMMY_SERVICE_ACCOUNT = None
 
 
@@ -728,6 +741,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"]