feat: add CNAME-based hostname support for V4 signed URLs

google/cloud/storage/bucket.py

@@ -513,33 +513,6 @@ def __init__(self, client, name=None, user_project=None):
  self._label_removals = set()
  self._user_project = user_project
 
- def __repr__(self):
- return "<Bucket: %s>" % (self.name,)
-
- @property
- def client(self):
- """The client bound to this bucket."""
- return self._client
-
- def _set_properties(self, value):
- """Set the properties for the current object.
-
- :type value: dict or :class:`google.cloud.storage.batch._FutureDict`
- :param value: The properties to be set.
- """
- self._label_removals.clear()
- return super(Bucket, self)._set_properties(value)
-
- @property
- def user_project(self):
- """Project ID to be billed for API requests made via this bucket.
-
- If unset, API requests are billed to the bucket owner.
-
- :rtype: str
- """
- return self._user_project
-
  @classmethod
  def from_string(cls, uri, client=None):
  """Get a constructor for bucket object by URI.
@@ -569,6 +542,33 @@ def from_string(cls, uri, client=None):
 
  return cls(client, name=netloc)
 
+ def __repr__(self):
+ return "<Bucket: %s>" % (self.name,)
+
+ @property
+ def client(self):
+ """The client bound to this bucket."""
+ return self._client
+
+ def _set_properties(self, value):
+ """Set the properties for the current object.
+
+ :type value: dict or :class:`google.cloud.storage.batch._FutureDict`
+ :param value: The properties to be set.
+ """
+ self._label_removals.clear()
+ return super(Bucket, self)._set_properties(value)
+
+ @property
+ def user_project(self):
+ """Project ID to be billed for API requests made via this bucket.
+
+ If unset, API requests are billed to the bucket owner.
+
+ :rtype: str
+ """
+ return self._user_project
+
  def blob(
  self,
  blob_name,
@@ -2355,6 +2355,8 @@ def generate_signed_url(
  credentials=None,
  version=None,
  virtual_hosted_style=False,
+ use_cname=None,
+ cname_scheme="http",
  ):
  """Generates a signed URL for this bucket.
 
@@ -2422,6 +2424,17 @@ def generate_signed_url(
  (Optional) If true, then construct the URL relative the bucket's
  virtual hostname, e.g., '<bucket-name>.storage.googleapis.com'.
 
+ :type use_cname: str
+ :param use_cname:
+ (Optional) If passed, then construct the URL relative to the value
+ as a CNAME. Value can be a bare hostname, e.g. "foo.bar.tld",
+ or a URL w/ scheme, e.g., "https://foo.bar.tld".
+
+ :type cname_scheme: str
+ :param cname_scheme:
+ (Optional) If ``use_cname`` is passed as a bare hostname, use
+ this value as the scheme. Defaults to ``"http"``.
+
  :raises: :exc:`ValueError` when version is invalid.
  :raises: :exc:`TypeError` when expiration is not a valid type.
  :raises: :exc:`AttributeError` if credentials is not an instance
@@ -2441,6 +2454,12 @@ def generate_signed_url(
  bucket_name=self.name
  )
  resource = "/"
+ elif use_cname is not None:
+ if ":" in use_cname:
+ api_access_endpoint = use_cname
+ else:
+ api_access_endpoint = "{}://{}".format(cname_scheme, use_cname)
+ resource = "/"
  else:
  resource = "/{bucket_name}".format(bucket_name=self.name)
 

tests/unit/test_bucket.py

@@ -415,6 +415,27 @@ def test_ctor_w_user_project(self):
  self.assertEqual(list(bucket._label_removals), [])
  self.assertEqual(bucket.user_project, USER_PROJECT)
 
+ def test_from_string_w_invalid_uri(self):
+ from google.cloud.storage.bucket import Bucket
+
+ connection = _Connection()
+ client = _Client(connection)
+
+ with pytest.raises(ValueError, match="URI scheme must be gs"):
+ Bucket.from_string("http://bucket_name", client)
+
+ def test_from_string_w_domain_name_bucket(self):
+ from google.cloud.storage.bucket import Bucket
+
+ connection = _Connection()
+ client = _Client(connection)
+ BUCKET_NAME = "buckets.example.com"
+ uri = "gs://" + BUCKET_NAME
+ bucket = Bucket.from_string(uri, client)
+ self.assertIsInstance(bucket, Bucket)
+ self.assertIs(bucket.client, client)
+ self.assertEqual(bucket.name, BUCKET_NAME)
+
  def test_blob_wo_keys(self):
  from google.cloud.storage.blob import Blob
 
@@ -1827,19 +1848,15 @@ def test_create_deprecated(self, mock_warn):
  )
 
  def test_create_w_user_project(self):
- from google.cloud.storage.client import Client
-
- PROJECT = "PROJECT"
  BUCKET_NAME = "bucket-name"
- DATA = {"name": BUCKET_NAME}
- connection = _make_connection(DATA)
- client = Client(project=PROJECT)
- client._base_connection = connection
-
+ connection = _Connection()
+ client = _Client(connection)
  bucket = self._make_one(client=client, name=BUCKET_NAME)
  bucket._user_project = "USER_PROJECT"
+
  with self.assertRaises(ValueError):
- bucket.create()
+ with mock.patch("warnings.warn"):
+ bucket.create()
 
  def test_versioning_enabled_setter(self):
  NAME = "name"
@@ -2740,6 +2757,8 @@ def _generate_signed_url_helper(
  credentials=None,
  expiration=None,
  virtual_hosted_style=False,
+ use_cname=None,
+ cname_scheme="http",
  ):
  from six.moves.urllib import parse
  from google.cloud._helpers import UTC
@@ -2775,6 +2794,8 @@ def _generate_signed_url_helper(
  query_parameters=query_parameters,
  version=version,
  virtual_hosted_style=virtual_hosted_style,
+ use_cname=use_cname,
+ cname_scheme=cname_scheme,
  )
 
  self.assertEqual(signed_uri, signer.return_value)
@@ -2789,6 +2810,12 @@ def _generate_signed_url_helper(
  bucket_name
  )
  expected_resource = "/"
+ elif use_cname is not None:
+ if ":" in use_cname:
+ expected_api_access_endpoint = use_cname
+ else:
+ expected_api_access_endpoint = "{}://{}".format(cname_scheme, use_cname)
+ expected_resource = "/"
  else:
  expected_api_access_endpoint = api_access_endpoint
  expected_resource = "/{}".format(parse.quote(bucket_name))
@@ -2815,27 +2842,6 @@ def test_get_bucket_from_string_w_valid_uri(self):
  self.assertIs(bucket.client, client)
  self.assertEqual(bucket.name, BUCKET_NAME)
 
- def test_get_bucket_from_string_w_invalid_uri(self):
- from google.cloud.storage.bucket import Bucket
-
- connection = _Connection()
- client = _Client(connection)
-
- with pytest.raises(ValueError, match="URI scheme must be gs"):
- Bucket.from_string("http://bucket_name", client)
-
- def test_get_bucket_from_string_w_domain_name_bucket(self):
- from google.cloud.storage.bucket import Bucket
-
- connection = _Connection()
- client = _Client(connection)
- BUCKET_NAME = "buckets.example.com"
- uri = "gs://" + BUCKET_NAME
- bucket = Bucket.from_string(uri, client)
- self.assertIsInstance(bucket, Bucket)
- self.assertIs(bucket.client, client)
- self.assertEqual(bucket.name, BUCKET_NAME)
-
  def test_generate_signed_url_no_version_passed_warning(self):
  self._generate_signed_url_helper()
 
@@ -2928,6 +2934,17 @@ def test_generate_signed_url_v4_w_credentials(self):
  def test_generate_signed_url_v4_w_virtual_hostname(self):
  self._generate_signed_url_v4_helper(virtual_hosted_style=True)
 
+ def test_generate_signed_url_v4_w_cname_wo_scheme(self):
+ self._generate_signed_url_v4_helper(use_cname="foo.bar.tld")
+
+ def test_generate_signed_url_v4_w_cname_w_scheme(self):
+ self._generate_signed_url_v4_helper(
+ use_cname="foo.bar.tld", cname_scheme="https"
+ )
+
+ def test_generate_signed_url_v4_w_cname_w_embedded_scheme(self):
+ self._generate_signed_url_v4_helper(use_cname="https://foo.bar.tld")
+
 
 class _Connection(object):
  _delete_bucket = False