Repository: googleapis/google-resumable-media-python

Issue:
Using a signed Resumable Upload URL fails with 403 unless content_type is empty
#### Environment details

  - Programming language: Python
  - OS: Ubuntu 20.04 in WSL2
  - Language runtime version: 3.8
  - Package version: cee416449701b72e7fd532585a1f739b02b6ab32

#### Steps to reproduce

  1. Generate a signed resumable upload url:
```
gsutil signurl -m RESUMABLE -c application/octet-stream -d 24h <key_file> gs://<bucket>/<obj>
```
  2. Run this python program, passing in the upload_url that was generated:
``` python
import io

from google.resumable_media import InvalidResponse
from google.resumable_media.requests import ResumableUpload
from requests import Session

chunk_size = 16 * 1024 * 1024  # 16MB

upload = ResumableUpload(upload_url, chunk_size)
stream = io.BytesIO(data)
metadata = {}
transport = Session()
transport.headers.update({'x-goog-resumable': 'start'})
try:
	response = upload.initiate(
		transport, stream, metadata, 'application/octet-stream')
except InvalidResponse as err:
	print(err.response.content)
	raise
print(response)
while not upload.finished:
	upload.transmit_next_chunk(transport)
stream.close()
```

You will get a 403 error from google that says `The request signature we calculated does not match the signature you provided. Check your Google secret key and signing method.`.  This is because the content_type passed by the library to the signed url is not correct.  If you instead generate the url without a content_type - it will succeed - but the object will have the content type `application/json`.  I'll send in a PR shortly with a fix that works for me.

Discussion:
williambrode: I'm sure my code is not production-worthy yet but I wanted to create the PR to help show what the problem is from my debugging.
williambrode: I signed the CLA.
williambrode: This doesn't handle v2 signed urls because they use a different query param ("Signature" I believe).  But I'm not sure that's a good reason to not support V4 signed urls.  We can add this but I don't have a test for it.
tseaver: @williambrode Thanks for the patch!

Can you double-check that the e-mail you used to create the commit (`Author: <wil*****de​@activision.com>`) matches the email(s) associated with your signed CLA?
williambrode: @tseaver You are right my e-mail was wrong in the commit.  I've fixed it.
Breathtender: @williambrode just wanted to follow up with you; are you able to write unit tests for this or would you like some support from us on that aspect. I know our testing framework might be a bit of a learning curve if this is your first external contribution :)
williambrode: Ah, I didn't quite realize that was directed at me.  I can take a look at adding the test.
Breathtender: Haha, reading over Tres' message again, it might not have been, I think I was hasty in my pass-through.
We can definitely handle getting a unit test written unless you'd like to give that a shot.
Breathtender: Alright, gave a go at the unit test. How's that look @tseaver ?
cojenco: Updated the unit tests to cover signed url uploads. @tseaver PTAL, thanks!

Patch:
diff --git a/google/resumable_media/_upload.py b/google/resumable_media/_upload.py
index 9cf4020..dc22cd1 100644
--- a/google/resumable_media/_upload.py
+++ b/google/resumable_media/_upload.py
@@ -27,6 +27,7 @@ import os
 import random
 import re
 import sys
+import urllib.parse
 
 from google import resumable_media
 from google.resumable_media import _helpers
@@ -462,10 +463,17 @@ class ResumableUpload(UploadBase):
 
         self._stream = stream
         self._content_type = content_type
-        headers = {
-            _CONTENT_TYPE_HEADER: "application/json; charset=UTF-8",
-            "x-upload-content-type": content_type,
-        }
+
+        # Signed URL requires content type set directly - not through x-upload-content-type
+        parse_result = urllib.parse.urlparse(self.upload_url)
+        parsed_query = urllib.parse.parse_qs(parse_result.query)
+        if "x-goog-signature" in parsed_query or "X-Goog-Signature" in parsed_query:
+            headers = {_CONTENT_TYPE_HEADER: content_type}
+        else:
+            headers = {
+                _CONTENT_TYPE_HEADER: "application/json; charset=UTF-8",
+                "x-upload-content-type": content_type,
+            }
         # Set the total bytes if possible.
         if total_bytes is not None:
             self._total_bytes = total_bytes


Test patch:
diff --git a/tests/unit/test__upload.py b/tests/unit/test__upload.py
index 2c2ca92..4e948f0 100644
--- a/tests/unit/test__upload.py
+++ b/tests/unit/test__upload.py
@@ -393,12 +393,14 @@ class TestResumableUpload(object):
         upload._total_bytes = 8192
         assert upload.total_bytes == 8192
 
-    def _prepare_initiate_request_helper(self, upload_headers=None, **method_kwargs):
+    def _prepare_initiate_request_helper(
+        self, upload_url=RESUMABLE_URL, upload_headers=None, **method_kwargs
+    ):
         data = b"some really big big data."
         stream = io.BytesIO(data)
         metadata = {"name": "big-data-file.txt"}
 
-        upload = _upload.ResumableUpload(RESUMABLE_URL, ONE_MB, headers=upload_headers)
+        upload = _upload.ResumableUpload(upload_url, ONE_MB, headers=upload_headers)
         orig_headers = upload._headers.copy()
         # Check ``upload``-s state before.
         assert upload._stream is None
@@ -435,6 +437,21 @@ class TestResumableUpload(object):
         }
         assert headers == expected_headers
 
+    def test_prepare_initiate_request_with_signed_url(self):
+        signed_urls = [
+            "https://storage.googleapis.com/b/o?x-goog-signature=123abc",
+            "https://storage.googleapis.com/b/o?X-Goog-Signature=123abc",
+        ]
+        for signed_url in signed_urls:
+            data, headers = self._prepare_initiate_request_helper(
+                upload_url=signed_url,
+            )
+            expected_headers = {
+                "content-type": BASIC_CONTENT,
+                "x-upload-content-length": "{:d}".format(len(data)),
+            }
+            assert headers == expected_headers
+
     def test__prepare_initiate_request_with_headers(self):
         headers = {"caviar": "beluga", "top": "quark"}
         data, new_headers = self._prepare_initiate_request_helper(