Help with signed URLs

[strayer]

Hi!

I'm trying to utilize gcp-storage-emulator for automated testing in https://github.com/strayer/django-gcloud-storage.

The basics are working, but two tests that are calling bucket.get_blob(name).generate_signed_url have issues… I built a fake credentials object that pretends to be able to sign:

from google.auth.credentials import AnonymousCredentials, Signing


class FakeSigningCredentials(Signing, AnonymousCredentials):
    def sign_bytes(self, message):
        return b"foobar"

    @property
    def signer_email(self):
        return "foobar@example.tld"

    @property
    def signer(self):
        pass

This makes the generate_signed_url function work in general. The returned URLs are problematic in two ways:

1. the host seems to be hardcoded to https://storage.googleapis.com in the SDK itself [see 1] – nothing gcp-storage-emulator can to about that

2. the path is not generated in a way gcp-storage emulator expects: GET /download/storage/v1/b/test_bucket_zdkgfs/test.jpg?Expires=1643405203&GoogleAccessId=foobar%40example.tld&Signature=Zm9vYmFy HTTP/1.1" 404 -

What is missing here is /o/ before the filename.

I've got my tests working by a very ugly hack to at least confirm some kind of working state by hacking the URLs returned by generate_signed_url:

        url = storage.url(file_name)
        if os.getenv('STORAGE_EMULATOR_HOST'):
            url = url.replace(f"/{file_name}", f"/o/{file_name}")
            url = url.replace("https://storage.googleapis.com", f"{os.getenv('STORAGE_EMULATOR_HOST')}/download/storage/v1/b")

        assert "image/jpeg" == urlopen(url).info().get("Content-Type")

Even though this is really ugly I can at least finally get automated tests without actual GCS access working, but I'm curious if I'm just missing something obvious here. Any help would be appreciated!

1: https://github.com/googleapis/python-storage/blob/main/google/cloud/storage/blob.py#L420

[oittaa]

I started investigating this a bit. I was testing with the Python package google-cloud-storage==2.1.0 and these code snippets https://cloud.google.com/storage/docs/access-control/signing-urls-with-helpers#storage-signed-url-object-python

The legacy version "v2" (or if the version is omitted) urls blob.generate_signed_url(version="v2", ...) look something like this:

https://storage.googleapis.com/my-bucket/my-blob?Expires=1643408524&GoogleAccessId=...

The newer "v4" urls blob.generate_signed_url(version="v4", ...) look something like this:

https://storage.googleapis.com/my-bucket/my-blob?X-Goog-Algorithm=GOOG4-RSA-SHA256&X-Goog-Credential=...

The version shouldn't matter with the emulator since we're just ignoring those parameters at the moment.

Looks like the signed urls are used only from the XML API, not from the JSON API. Those need to be added to the the supported URL endpoints, but that shouldn't be too time consuming. https://cloud.google.com/storage/docs/request-endpoints#xml-api

Back to your first point. Could you pass api_access_endpoint parameter if STORAGE_EMULATOR_HOST environment variable exists? That should solve the issue. Of course it would be nicer if the official Google library would detect STORAGE_EMULATOR_HOST and use it automatically. https://googleapis.dev/python/storage/latest/blobs.html#google.cloud.storage.blob.Blob.generate_signed_url

[oittaa]

Uploading with signed URLs is supported now.

gcp-storage-emulator/tests/test_server.py

Lines 797 to 838 in 646ea0c

	def test_signed_url_download(self):
	content = b"The quick brown fox jumps over the lazy dog"
	bucket = self._client.create_bucket("testbucket")
	blob = bucket.blob("signed-download")
	blob.upload_from_string(content)
	url = blob.generate_signed_url(
	api_access_endpoint="http://localhost:9023",
	credentials=FakeSigningCredentials(),
	version="v4",
	expiration=datetime.timedelta(minutes=15),
	method="GET",
	)
	response = requests.get(url)
	self.assertEqual(response.content, content)
	def test_signed_url_upload(self):
	test_text = os.path.join(
	os.path.dirname(os.path.abspath(__file__)), "test_text.txt"
	)
	bucket = self._client.create_bucket("testbucket")
	blob = bucket.blob("signed-upload")
	url = blob.generate_signed_url(
	api_access_endpoint="http://localhost:9023",
	credentials=FakeSigningCredentials(),
	version="v4",
	expiration=datetime.timedelta(minutes=15),
	method="PUT",
	)
	with open(test_text, "rb") as file:
	headers = {"Content-type": "text/plain"}
	response = requests.put(url, data=file, headers=headers)
	self.assertEqual(response.status_code, 200)
	blob_content = blob.download_as_bytes()
	file.seek(0)
	self.assertEqual(blob_content, file.read())
	self.assertEqual(blob.content_type, "text/plain")

[oittaa]

I took a quick look at your code. Maybe you could use something like this, which is very easy to test by just passing the endpoint as an argument.

    def url(self, name, api_access_endpoint="https://storage.googleapis.com"):
        name = safe_join(self.bucket_subdir, name)
        name = prepare_name(name)

        if self.use_unsigned_urls:
          return "{}/{}/{}".format(api_access_endpoint, self.bucket.name, name)

        return self.bucket.get_blob(name).generate_signed_url(api_access_endpoint=api_access_endpoint, expiration=datetime.datetime.now() + datetime.timedelta(hours=1))

Or maybe even as an argument to DjangoGCloudStorage class...

[strayer]

Thank you so much for looking into this so quickly! Also thanks for extending the code to support the XML API - fun to see the quick hack I did for signing credentials here :D

I will most likely do something comparable to what you proposed for my url function, already thought about adding an optional constructor parameter like you mentioned.

I'll report back when I had time to try the new version.

[oittaa]

I'm going to close this, but feel free to open another ticket if you encounter other issues.