diff --git a/google/cloud/storage/_helpers.py b/google/cloud/storage/_helpers.py
index 0fb4e0ff8..7afa43d7d 100644
--- a/google/cloud/storage/_helpers.py
+++ b/google/cloud/storage/_helpers.py
@@ -21,6 +21,7 @@
 from hashlib import md5
 import os
 from urllib.parse import urlsplit
+from urllib.parse import urlunsplit
 from uuid import uuid4
 
 from google import resumable_media
@@ -30,19 +31,24 @@
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
 
 
-STORAGE_EMULATOR_ENV_VAR = "STORAGE_EMULATOR_HOST"
+STORAGE_EMULATOR_ENV_VAR = "STORAGE_EMULATOR_HOST"  # Despite name, includes scheme.
 """Environment variable defining host for Storage emulator."""
 
-_API_ENDPOINT_OVERRIDE_ENV_VAR = "API_ENDPOINT_OVERRIDE"
+_API_ENDPOINT_OVERRIDE_ENV_VAR = "API_ENDPOINT_OVERRIDE"  # Includes scheme.
 """This is an experimental configuration variable. Use api_endpoint instead."""
 
 _API_VERSION_OVERRIDE_ENV_VAR = "API_VERSION_OVERRIDE"
 """This is an experimental configuration variable used for internal testing."""
 
-_DEFAULT_STORAGE_HOST = os.getenv(
-    _API_ENDPOINT_OVERRIDE_ENV_VAR, "https://storage.googleapis.com"
+_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
+
+_STORAGE_HOST_TEMPLATE = "storage.{universe_domain}"
+
+_TRUE_DEFAULT_STORAGE_HOST = _STORAGE_HOST_TEMPLATE.format(
+    universe_domain=_DEFAULT_UNIVERSE_DOMAIN
 )
-"""Default storage host for JSON API."""
+
+_DEFAULT_SCHEME = "https://"
 
 _API_VERSION = os.getenv(_API_VERSION_OVERRIDE_ENV_VAR, "v1")
 """API version of the default storage host"""
@@ -72,8 +78,39 @@
 )
 
 
-def _get_storage_host():
-    return os.environ.get(STORAGE_EMULATOR_ENV_VAR, _DEFAULT_STORAGE_HOST)
+def _get_storage_emulator_override():
+    return os.environ.get(STORAGE_EMULATOR_ENV_VAR, None)
+
+
+def _get_default_storage_base_url():
+    return os.getenv(
+        _API_ENDPOINT_OVERRIDE_ENV_VAR, _DEFAULT_SCHEME + _TRUE_DEFAULT_STORAGE_HOST
+    )
+
+
+def _get_api_endpoint_override():
+    """This is an experimental configuration variable. Use api_endpoint instead."""
+    if _get_default_storage_base_url() != _DEFAULT_SCHEME + _TRUE_DEFAULT_STORAGE_HOST:
+        return _get_default_storage_base_url()
+    return None
+
+
+def _virtual_hosted_style_base_url(url, bucket, trailing_slash=False):
+    """Returns the scheme and netloc sections of the url, with the bucket
+    prepended to the netloc.
+
+    Not intended for use with netlocs which include a username and password.
+    """
+    parsed_url = urlsplit(url)
+    new_netloc = f"{bucket}.{parsed_url.netloc}"
+    base_url = urlunsplit(
+        (parsed_url.scheme, new_netloc, "/" if trailing_slash else "", "", "")
+    )
+    return base_url
+
+
+def _use_client_cert():
+    return os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE") == "true"
 
 
 def _get_environ_project():
diff --git a/google/cloud/storage/_http.py b/google/cloud/storage/_http.py
index fdf1d56b4..b4e16ebe4 100644
--- a/google/cloud/storage/_http.py
+++ b/google/cloud/storage/_http.py
@@ -21,8 +21,14 @@
 
 
 class Connection(_http.JSONConnection):
-    """A connection to Google Cloud Storage via the JSON REST API. Mutual TLS feature will be
-    enabled if `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "true".
+    """A connection to Google Cloud Storage via the JSON REST API.
+
+    Mutual TLS will be enabled if the "GOOGLE_API_USE_CLIENT_CERTIFICATE"
+    environment variable is set to the exact string "true" (case-sensitive).
+
+    Mutual TLS is not compatible with any API endpoint or universe domain
+    override at this time. If such settings are enabled along with
+    "GOOGLE_API_USE_CLIENT_CERTIFICATE", a ValueError will be raised.
 
     :type client: :class:`~google.cloud.storage.client.Client`
     :param client: The client that owns the current connection.
@@ -34,7 +40,7 @@ class Connection(_http.JSONConnection):
     :param api_endpoint: (Optional) api endpoint to use.
     """
 
-    DEFAULT_API_ENDPOINT = _helpers._DEFAULT_STORAGE_HOST
+    DEFAULT_API_ENDPOINT = _helpers._get_default_storage_base_url()
     DEFAULT_API_MTLS_ENDPOINT = "https://storage.mtls.googleapis.com"
 
     def __init__(self, client, client_info=None, api_endpoint=None):
diff --git a/google/cloud/storage/_signing.py b/google/cloud/storage/_signing.py
index 1ec61142d..10232ac51 100644
--- a/google/cloud/storage/_signing.py
+++ b/google/cloud/storage/_signing.py
@@ -466,7 +466,7 @@ def generate_signed_url_v4(
                        ``tzinfo`` set,  it will be assumed to be ``UTC``.
 
     :type api_access_endpoint: str
-    :param api_access_endpoint: (Optional) URI base. Defaults to
+    :param api_access_endpoint: URI base. Defaults to
                                 "https://storage.googleapis.com/"
 
     :type method: str
diff --git a/google/cloud/storage/blob.py b/google/cloud/storage/blob.py
index 47564b6da..6cfa56190 100644
--- a/google/cloud/storage/blob.py
+++ b/google/cloud/storage/blob.py
@@ -57,11 +57,12 @@
 from google.cloud.storage._helpers import _raise_if_more_than_one_set
 from google.cloud.storage._helpers import _api_core_retry_to_resumable_media_retry
 from google.cloud.storage._helpers import _get_default_headers
+from google.cloud.storage._helpers import _get_default_storage_base_url
 from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
 from google.cloud.storage._helpers import _NUM_RETRIES_MESSAGE
-from google.cloud.storage._helpers import _DEFAULT_STORAGE_HOST
 from google.cloud.storage._helpers import _API_VERSION
+from google.cloud.storage._helpers import _virtual_hosted_style_base_url
 from google.cloud.storage.acl import ACL
 from google.cloud.storage.acl import ObjectACL
 from google.cloud.storage.constants import _DEFAULT_TIMEOUT
@@ -80,7 +81,6 @@
 from google.cloud.storage.fileio import BlobWriter
 
 
-_API_ACCESS_ENDPOINT = _DEFAULT_STORAGE_HOST
 _DEFAULT_CONTENT_TYPE = "application/octet-stream"
 _DOWNLOAD_URL_TEMPLATE = "{hostname}/download/storage/{api_version}{path}?alt=media"
 _BASE_UPLOAD_TEMPLATE = (
@@ -376,8 +376,12 @@ def public_url(self):
         :rtype: `string`
         :returns: The public URL for this blob.
         """
+        if self.client:
+            endpoint = self.client.api_endpoint
+        else:
+            endpoint = _get_default_storage_base_url()
         return "{storage_base_url}/{bucket_name}/{quoted_name}".format(
-            storage_base_url=_API_ACCESS_ENDPOINT,
+            storage_base_url=endpoint,
             bucket_name=self.bucket.name,
             quoted_name=_quote(self.name, safe=b"/~"),
         )
@@ -416,7 +420,7 @@ def from_string(cls, uri, client=None):
     def generate_signed_url(
         self,
         expiration=None,
-        api_access_endpoint=_API_ACCESS_ENDPOINT,
+        api_access_endpoint=None,
         method="GET",
         content_md5=None,
         content_type=None,
@@ -464,7 +468,9 @@ def generate_signed_url(
             assumed to be ``UTC``.
 
         :type api_access_endpoint: str
-        :param api_access_endpoint: (Optional) URI base.
+        :param api_access_endpoint: (Optional) URI base, for instance
+            "https://storage.googleapis.com". If not specified, the client's
+            api_endpoint will be used. Incompatible with bucket_bound_hostname.
 
         :type method: str
         :param method: The HTTP verb that will be used when requesting the URL.
@@ -537,13 +543,14 @@ def generate_signed_url(
         :param virtual_hosted_style:
             (Optional) If true, then construct the URL relative the bucket's
             virtual hostname, e.g., '<bucket-name>.storage.googleapis.com'.
+            Incompatible with bucket_bound_hostname.
 
         :type bucket_bound_hostname: str
         :param bucket_bound_hostname:
-            (Optional) If passed, then construct the URL relative to the
-            bucket-bound hostname.  Value can be a bare or with scheme, e.g.,
-            'example.com' or 'http://example.com'.  See:
-            https://cloud.google.com/storage/docs/request-endpoints#cname
+            (Optional) If passed, then construct the URL relative to the bucket-bound hostname.
+            Value can be a bare or with scheme, e.g., 'example.com' or 'http://example.com'.
+            Incompatible with api_access_endpoint and virtual_hosted_style.
+            See: https://cloud.google.com/storage/docs/request-endpoints#cname
 
         :type scheme: str
         :param scheme:
@@ -551,7 +558,7 @@ def generate_signed_url(
             hostname, use this value as the scheme.  ``https`` will work only
             when using a CDN.  Defaults to ``"http"``.
 
-        :raises: :exc:`ValueError` when version is invalid.
+        :raises: :exc:`ValueError` when version is invalid or mutually exclusive arguments are used.
         :raises: :exc:`TypeError` when expiration is not a valid type.
         :raises: :exc:`AttributeError` if credentials is not an instance
                 of :class:`google.auth.credentials.Signing`.
@@ -565,25 +572,38 @@ def generate_signed_url(
         elif version not in ("v2", "v4"):
             raise ValueError("'version' must be either 'v2' or 'v4'")
 
+        if (
+            api_access_endpoint is not None or virtual_hosted_style
+        ) and bucket_bound_hostname:
+            raise ValueError(
+                "The bucket_bound_hostname argument is not compatible with "
+                "either api_access_endpoint or virtual_hosted_style."
+            )
+
+        if api_access_endpoint is None:
+            client = self._require_client(client)
+            api_access_endpoint = client.api_endpoint
+
         quoted_name = _quote(self.name, safe=b"/~")
 
         # If you are on Google Compute Engine, you can't generate a signed URL
         # using GCE service account.
         # See https://github.com/googleapis/google-auth-library-python/issues/50
         if virtual_hosted_style:
-            api_access_endpoint = f"https://{self.bucket.name}.storage.googleapis.com"
+            api_access_endpoint = _virtual_hosted_style_base_url(
+                api_access_endpoint, self.bucket.name
+            )
+            resource = f"/{quoted_name}"
         elif bucket_bound_hostname:
             api_access_endpoint = _bucket_bound_hostname_url(
                 bucket_bound_hostname, scheme
             )
+            resource = f"/{quoted_name}"
         else:
             resource = f"/{self.bucket.name}/{quoted_name}"
 
-        if virtual_hosted_style or bucket_bound_hostname:
-            resource = f"/{quoted_name}"
-
         if credentials is None:
-            client = self._require_client(client)
+            client = self._require_client(client)  # May be redundant, but that's ok.
             credentials = client._credentials
 
         if version == "v2":
diff --git a/google/cloud/storage/bucket.py b/google/cloud/storage/bucket.py
index 95017a14d..8f44f07d3 100644
--- a/google/cloud/storage/bucket.py
+++ b/google/cloud/storage/bucket.py
@@ -36,6 +36,7 @@
 from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
 from google.cloud.storage._helpers import _bucket_bound_hostname_url
+from google.cloud.storage._helpers import _virtual_hosted_style_base_url
 from google.cloud.storage.acl import BucketACL
 from google.cloud.storage.acl import DefaultObjectACL
 from google.cloud.storage.blob import Blob
@@ -82,7 +83,6 @@
     "valid before the bucket is created. Instead, pass the location "
     "to `Bucket.create`."
 )
-_API_ACCESS_ENDPOINT = "https://storage.googleapis.com"
 
 
 def _blobs_page_start(iterator, page, response):
@@ -3265,7 +3265,7 @@ def lock_retention_policy(
     def generate_signed_url(
         self,
         expiration=None,
-        api_access_endpoint=_API_ACCESS_ENDPOINT,
+        api_access_endpoint=None,
         method="GET",
         headers=None,
         query_parameters=None,
@@ -3298,7 +3298,9 @@ def generate_signed_url(
                            ``tzinfo`` set,  it will be assumed to be ``UTC``.
 
         :type api_access_endpoint: str
-        :param api_access_endpoint: (Optional) URI base.
+        :param api_access_endpoint: (Optional) URI base, for instance
+            "https://storage.googleapis.com". If not specified, the client's
+            api_endpoint will be used. Incompatible with bucket_bound_hostname.
 
         :type method: str
         :param method: The HTTP verb that will be used when requesting the URL.
@@ -3322,7 +3324,6 @@ def generate_signed_url(
         :param client: (Optional) The client to use.  If not passed, falls back
                        to the ``client`` stored on the blob's bucket.
 
-
         :type credentials: :class:`google.auth.credentials.Credentials` or
                            :class:`NoneType`
         :param credentials: The authorization credentials to attach to requests.
@@ -3338,11 +3339,13 @@ def generate_signed_url(
         :param virtual_hosted_style:
             (Optional) If true, then construct the URL relative the bucket's
             virtual hostname, e.g., '<bucket-name>.storage.googleapis.com'.
+            Incompatible with bucket_bound_hostname.
 
         :type bucket_bound_hostname: str
         :param bucket_bound_hostname:
-            (Optional) If pass, then construct the URL relative to the bucket-bound hostname.
-            Value cane be a bare or with scheme, e.g., 'example.com' or 'http://example.com'.
+            (Optional) If passed, then construct the URL relative to the bucket-bound hostname.
+            Value can be a bare or with scheme, e.g., 'example.com' or 'http://example.com'.
+            Incompatible with api_access_endpoint and virtual_hosted_style.
             See: https://cloud.google.com/storage/docs/request-endpoints#cname
 
         :type scheme: str
@@ -3351,7 +3354,7 @@ def generate_signed_url(
             this value as the scheme.  ``https`` will work only when using a CDN.
             Defaults to ``"http"``.
 
-        :raises: :exc:`ValueError` when version is invalid.
+        :raises: :exc:`ValueError` when version is invalid or mutually exclusive arguments are used.
         :raises: :exc:`TypeError` when expiration is not a valid type.
         :raises: :exc:`AttributeError` if credentials is not an instance
                 of :class:`google.auth.credentials.Signing`.
@@ -3365,23 +3368,36 @@ def generate_signed_url(
         elif version not in ("v2", "v4"):
             raise ValueError("'version' must be either 'v2' or 'v4'")
 
+        if (
+            api_access_endpoint is not None or virtual_hosted_style
+        ) and bucket_bound_hostname:
+            raise ValueError(
+                "The bucket_bound_hostname argument is not compatible with "
+                "either api_access_endpoint or virtual_hosted_style."
+            )
+
+        if api_access_endpoint is None:
+            client = self._require_client(client)
+            api_access_endpoint = client.api_endpoint
+
         # If you are on Google Compute Engine, you can't generate a signed URL
         # using GCE service account.
         # See https://github.com/googleapis/google-auth-library-python/issues/50
         if virtual_hosted_style:
-            api_access_endpoint = f"https://{self.name}.storage.googleapis.com"
+            api_access_endpoint = _virtual_hosted_style_base_url(
+                api_access_endpoint, self.name
+            )
+            resource = "/"
         elif bucket_bound_hostname:
             api_access_endpoint = _bucket_bound_hostname_url(
                 bucket_bound_hostname, scheme
             )
+            resource = "/"
         else:
             resource = f"/{self.name}"
 
-        if virtual_hosted_style or bucket_bound_hostname:
-            resource = "/"
-
         if credentials is None:
-            client = self._require_client(client)
+            client = self._require_client(client)  # May be redundant, but that's ok.
             credentials = client._credentials
 
         if version == "v2":
diff --git a/google/cloud/storage/client.py b/google/cloud/storage/client.py
index 69019f218..2fcdaecd8 100644
--- a/google/cloud/storage/client.py
+++ b/google/cloud/storage/client.py
@@ -31,9 +31,14 @@
 from google.cloud.exceptions import NotFound
 
 from google.cloud.storage._helpers import _get_environ_project
-from google.cloud.storage._helpers import _get_storage_host
-from google.cloud.storage._helpers import _DEFAULT_STORAGE_HOST
+from google.cloud.storage._helpers import _use_client_cert
+from google.cloud.storage._helpers import _get_storage_emulator_override
+from google.cloud.storage._helpers import _get_api_endpoint_override
+from google.cloud.storage._helpers import _STORAGE_HOST_TEMPLATE
 from google.cloud.storage._helpers import _bucket_bound_hostname_url
+from google.cloud.storage._helpers import _DEFAULT_UNIVERSE_DOMAIN
+from google.cloud.storage._helpers import _DEFAULT_SCHEME
+from google.cloud.storage._helpers import _virtual_hosted_style_base_url
 
 from google.cloud.storage._http import Connection
 from google.cloud.storage._signing import (
@@ -87,7 +92,7 @@ class Client(ClientWithProject):
 
     :type client_options: :class:`~google.api_core.client_options.ClientOptions` or :class:`dict`
     :param client_options: (Optional) Client options used to set user options on the client.
-        API Endpoint should be set through client_options.
+        A non-default universe domain or api endpoint should be set through client_options.
 
     :type use_auth_w_custom_endpoint: bool
     :param use_auth_w_custom_endpoint:
@@ -135,32 +140,79 @@ def __init__(
         self._initial_client_options = client_options
         self._extra_headers = extra_headers
 
-        kw_args = {"client_info": client_info}
-
-        # `api_endpoint` should be only set by the user via `client_options`,
-        # or if the _get_storage_host() returns a non-default value (_is_emulator_set).
-        # `api_endpoint` plays an important role for mTLS, if it is not set,
-        # then mTLS logic will be applied to decide which endpoint will be used.
-        storage_host = _get_storage_host()
-        _is_emulator_set = storage_host != _DEFAULT_STORAGE_HOST
-        kw_args["api_endpoint"] = storage_host if _is_emulator_set else None
+        connection_kw_args = {"client_info": client_info}
 
         if client_options:
             if isinstance(client_options, dict):
                 client_options = google.api_core.client_options.from_dict(
                     client_options
                 )
-            if client_options.api_endpoint:
-                api_endpoint = client_options.api_endpoint
-                kw_args["api_endpoint"] = api_endpoint
+
+        if client_options and client_options.universe_domain:
+            self._universe_domain = client_options.universe_domain
+        else:
+            self._universe_domain = None
+
+        storage_emulator_override = _get_storage_emulator_override()
+        api_endpoint_override = _get_api_endpoint_override()
+
+        # Determine the api endpoint. The rules are as follows:
+
+        # 1. If the `api_endpoint` is set in `client_options`, use that as the
+        #    endpoint.
+        if client_options and client_options.api_endpoint:
+            api_endpoint = client_options.api_endpoint
+
+        # 2. Elif the "STORAGE_EMULATOR_HOST" env var is set, then use that as the
+        #    endpoint.
+        elif storage_emulator_override:
+            api_endpoint = storage_emulator_override
+
+        # 3. Elif the "API_ENDPOINT_OVERRIDE" env var is set, then use that as the
+        #    endpoint.
+        elif api_endpoint_override:
+            api_endpoint = api_endpoint_override
+
+        # 4. Elif the `universe_domain` is set in `client_options`,
+        #    create the endpoint using that as the default.
+        #
+        #    Mutual TLS is not compatible with a non-default universe domain
+        #    at this time. If such settings are enabled along with the
+        #    "GOOGLE_API_USE_CLIENT_CERTIFICATE" env variable, a ValueError will
+        #    be raised.
+
+        elif self._universe_domain:
+            # The final decision of whether to use mTLS takes place in
+            # google-auth-library-python. We peek at the environment variable
+            # here only to issue an exception in case of a conflict.
+            if _use_client_cert():
+                raise ValueError(
+                    'The "GOOGLE_API_USE_CLIENT_CERTIFICATE" env variable is '
+                    'set to "true" and a non-default universe domain is '
+                    "configured. mTLS is not supported in any universe other than"
+                    "googleapis.com."
+                )
+            api_endpoint = _DEFAULT_SCHEME + _STORAGE_HOST_TEMPLATE.format(
+                universe_domain=self._universe_domain
+            )
+
+        # 5. Else, use the default, which is to use the default
+        #    universe domain of "googleapis.com" and create the endpoint
+        #    "storage.googleapis.com" from that.
+        else:
+            api_endpoint = None
+
+        connection_kw_args["api_endpoint"] = api_endpoint
+
+        self._is_emulator_set = True if storage_emulator_override else False
 
         # If a custom endpoint is set, the client checks for credentials
         # or finds the default credentials based on the current environment.
         # Authentication may be bypassed under certain conditions:
         # (1) STORAGE_EMULATOR_HOST is set (for backwards compatibility), OR
         # (2) use_auth_w_custom_endpoint is set to False.
-        if kw_args["api_endpoint"] is not None:
-            if _is_emulator_set or not use_auth_w_custom_endpoint:
+        if connection_kw_args["api_endpoint"] is not None:
+            if self._is_emulator_set or not use_auth_w_custom_endpoint:
                 if credentials is None:
                     credentials = AnonymousCredentials()
                 if project is None:
@@ -176,11 +228,24 @@ def __init__(
             _http=_http,
         )
 
+        # Validate that the universe domain of the credentials matches the
+        # universe domain of the client.
+        if self._credentials.universe_domain != self.universe_domain:
+            raise ValueError(
+                "The configured universe domain ({client_ud}) does not match "
+                "the universe domain found in the credentials ({cred_ud}). If "
+                "you haven't configured the universe domain explicitly, "
+                "`googleapis.com` is the default.".format(
+                    client_ud=self.universe_domain,
+                    cred_ud=self._credentials.universe_domain,
+                )
+            )
+
         if no_project:
             self.project = None
 
         # Pass extra_headers to Connection
-        connection = Connection(self, **kw_args)
+        connection = Connection(self, **connection_kw_args)
         connection.extra_headers = extra_headers
         self._connection = connection
         self._batch_stack = _LocalStack()
@@ -201,6 +266,14 @@ def create_anonymous_client(cls):
         client.project = None
         return client
 
+    @property
+    def universe_domain(self):
+        return self._universe_domain or _DEFAULT_UNIVERSE_DOMAIN
+
+    @property
+    def api_endpoint(self):
+        return self._connection.API_BASE_URL
+
     @property
     def _connection(self):
         """Get connection or batch on the client.
@@ -922,8 +995,7 @@ def create_bucket(
             project = self.project
 
         # Use no project if STORAGE_EMULATOR_HOST is set
-        _is_emulator_set = _get_storage_host() != _DEFAULT_STORAGE_HOST
-        if _is_emulator_set:
+        if self._is_emulator_set:
             if project is None:
                 project = _get_environ_project()
             if project is None:
@@ -1338,8 +1410,7 @@ def list_buckets(
             project = self.project
 
         # Use no project if STORAGE_EMULATOR_HOST is set
-        _is_emulator_set = _get_storage_host() != _DEFAULT_STORAGE_HOST
-        if _is_emulator_set:
+        if self._is_emulator_set:
             if project is None:
                 project = _get_environ_project()
             if project is None:
@@ -1574,13 +1645,16 @@ def generate_signed_post_policy_v4(
                             key to sign text.
 
         :type virtual_hosted_style: bool
-        :param virtual_hosted_style: (Optional) If True, construct the URL relative to the bucket
-                                     virtual hostname, e.g., '<bucket-name>.storage.googleapis.com'.
+        :param virtual_hosted_style:
+            (Optional) If True, construct the URL relative to the bucket
+            virtual hostname, e.g., '<bucket-name>.storage.googleapis.com'.
+            Incompatible with bucket_bound_hostname.
 
         :type bucket_bound_hostname: str
         :param bucket_bound_hostname:
             (Optional) If passed, construct the URL relative to the bucket-bound hostname.
             Value can be bare or with a scheme, e.g., 'example.com' or 'http://example.com'.
+            Incompatible with virtual_hosted_style.
             See: https://cloud.google.com/storage/docs/request-endpoints#cname
 
         :type scheme: str
@@ -1595,9 +1669,17 @@ def generate_signed_post_policy_v4(
         :type access_token: str
         :param access_token: (Optional) Access token for a service account.
 
+        :raises: :exc:`ValueError` when mutually exclusive arguments are used.
+
         :rtype: dict
         :returns: Signed POST policy.
         """
+        if virtual_hosted_style and bucket_bound_hostname:
+            raise ValueError(
+                "Only one of virtual_hosted_style and bucket_bound_hostname "
+                "can be specified."
+            )
+
         credentials = self._credentials if credentials is None else credentials
         ensure_signed_credentials(credentials)
 
@@ -1669,11 +1751,13 @@ def generate_signed_post_policy_v4(
         )
         # designate URL
         if virtual_hosted_style:
-            url = f"https://{bucket_name}.storage.googleapis.com/"
+            url = _virtual_hosted_style_base_url(
+                self.api_endpoint, bucket_name, trailing_slash=True
+            )
         elif bucket_bound_hostname:
             url = f"{_bucket_bound_hostname_url(bucket_bound_hostname, scheme)}/"
         else:
-            url = f"https://storage.googleapis.com/{bucket_name}/"
+            url = f"{self.api_endpoint}/{bucket_name}/"
 
         return {"url": url, "fields": policy_fields}
 
diff --git a/setup.py b/setup.py
index fa0200cdf..b2f5e411e 100644
--- a/setup.py
+++ b/setup.py
@@ -28,8 +28,8 @@
 # 'Development Status :: 5 - Production/Stable'
 release_status = "Development Status :: 5 - Production/Stable"
 dependencies = [
-    "google-auth >= 2.23.3, < 3.0dev",
-    "google-api-core >= 1.31.5, <3.0.0dev,!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0",
+    "google-auth >= 2.26.1, < 3.0dev",
+    "google-api-core >= 2.15.0, <3.0.0dev",
     "google-cloud-core >= 2.3.0, < 3.0dev",
     "google-resumable-media >= 2.6.0",
     "requests >= 2.18.0, < 3.0.0dev",
diff --git a/tests/system/_helpers.py b/tests/system/_helpers.py
index e298d7932..a044c4ca8 100644
--- a/tests/system/_helpers.py
+++ b/tests/system/_helpers.py
@@ -20,7 +20,7 @@
 from test_utils.retry import RetryErrors
 from test_utils.retry import RetryInstanceState
 from test_utils.system import unique_resource_id
-from google.cloud.storage._helpers import _DEFAULT_STORAGE_HOST
+from google.cloud.storage._helpers import _get_default_storage_base_url
 
 retry_429 = RetryErrors(exceptions.TooManyRequests)
 retry_429_harder = RetryErrors(exceptions.TooManyRequests, max_tries=10)
@@ -32,7 +32,9 @@
 user_project = os.environ.get("GOOGLE_CLOUD_TESTS_USER_PROJECT")
 testing_mtls = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE") == "true"
 signing_blob_content = b"This time for sure, Rocky!"
-is_api_endpoint_override = _DEFAULT_STORAGE_HOST != "https://storage.googleapis.com"
+is_api_endpoint_override = (
+    _get_default_storage_base_url() != "https://storage.googleapis.com"
+)
 
 
 def _bad_copy(bad_request):
diff --git a/tests/system/test__signing.py b/tests/system/test__signing.py
index 26d73e543..30a898e11 100644
--- a/tests/system/test__signing.py
+++ b/tests/system/test__signing.py
@@ -45,7 +45,7 @@ def _create_signed_list_blobs_url_helper(
         method=method,
         client=client,
         version=version,
-        api_access_endpoint=_helpers._DEFAULT_STORAGE_HOST,
+        api_access_endpoint=_helpers._get_default_storage_base_url(),
     )
 
     response = requests.get(signed_url)
diff --git a/tests/unit/test__helpers.py b/tests/unit/test__helpers.py
index 7f05a8d00..401e0dd15 100644
--- a/tests/unit/test__helpers.py
+++ b/tests/unit/test__helpers.py
@@ -22,20 +22,18 @@
 GCCL_INVOCATION_TEST_CONST = "gccl-invocation-id/test-invocation-123"
 
 
-class Test__get_storage_host(unittest.TestCase):
+class Test__get_storage_emulator_override(unittest.TestCase):
     @staticmethod
     def _call_fut():
-        from google.cloud.storage._helpers import _get_storage_host
+        from google.cloud.storage._helpers import _get_storage_emulator_override
 
-        return _get_storage_host()
+        return _get_storage_emulator_override()
 
     def test_wo_env_var(self):
-        from google.cloud.storage._helpers import _DEFAULT_STORAGE_HOST
-
         with mock.patch("os.environ", {}):
-            host = self._call_fut()
+            override = self._call_fut()
 
-        self.assertEqual(host, _DEFAULT_STORAGE_HOST)
+        self.assertIsNone(override)
 
     def test_w_env_var(self):
         from google.cloud.storage._helpers import STORAGE_EMULATOR_ENV_VAR
@@ -43,9 +41,36 @@ def test_w_env_var(self):
         HOST = "https://api.example.com"
 
         with mock.patch("os.environ", {STORAGE_EMULATOR_ENV_VAR: HOST}):
-            host = self._call_fut()
+            emu = self._call_fut()
+
+        self.assertEqual(emu, HOST)
+
+
+class Test__get_api_endpoint_override(unittest.TestCase):
+    @staticmethod
+    def _call_fut():
+        from google.cloud.storage._helpers import _get_api_endpoint_override
+
+        return _get_api_endpoint_override()
+
+    def test_wo_env_var(self):
+        from google.cloud.storage._helpers import _TRUE_DEFAULT_STORAGE_HOST
+        from google.cloud.storage._helpers import _DEFAULT_SCHEME
+
+        with mock.patch("os.environ", {}):
+            override = self._call_fut()
+
+        self.assertIsNone(override, _DEFAULT_SCHEME + _TRUE_DEFAULT_STORAGE_HOST)
+
+    def test_w_env_var(self):
+        from google.cloud.storage._helpers import _API_ENDPOINT_OVERRIDE_ENV_VAR
+
+        BASE_URL = "https://api.example.com"
+
+        with mock.patch("os.environ", {_API_ENDPOINT_OVERRIDE_ENV_VAR: BASE_URL}):
+            override = self._call_fut()
 
-        self.assertEqual(host, HOST)
+        self.assertEqual(override, BASE_URL)
 
 
 class Test__get_environ_project(unittest.TestCase):
diff --git a/tests/unit/test__http.py b/tests/unit/test__http.py
index 3ea3ed1a4..33ff1a890 100644
--- a/tests/unit/test__http.py
+++ b/tests/unit/test__http.py
@@ -89,7 +89,10 @@ def test_metadata_op_has_client_custom_headers(self):
         response._content = data
         http.is_mtls = False
         http.request.return_value = response
-        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+        credentials = mock.Mock(
+            spec=google.auth.credentials.Credentials,
+            universe_domain=_helpers._DEFAULT_UNIVERSE_DOMAIN,
+        )
         client = Client(
             project="project",
             credentials=credentials,
diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py
index c1f6bad9a..3070af956 100644
--- a/tests/unit/test_batch.py
+++ b/tests/unit/test_batch.py
@@ -20,11 +20,16 @@
 import mock
 import requests
 
+from google.cloud.storage._helpers import _DEFAULT_UNIVERSE_DOMAIN
+
 
 def _make_credentials():
     import google.auth.credentials
 
-    return mock.Mock(spec=google.auth.credentials.Credentials)
+    return mock.Mock(
+        spec=google.auth.credentials.Credentials,
+        universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+    )
 
 
 def _make_response(status=http.client.OK, content=b"", headers={}):
diff --git a/tests/unit/test_blob.py b/tests/unit/test_blob.py
index 563111ef0..805dae741 100644
--- a/tests/unit/test_blob.py
+++ b/tests/unit/test_blob.py
@@ -29,6 +29,8 @@
 
 from google.cloud.storage import _helpers
 from google.cloud.storage._helpers import _get_default_headers
+from google.cloud.storage._helpers import _DEFAULT_UNIVERSE_DOMAIN
+from google.cloud.storage._helpers import _get_default_storage_base_url
 from google.cloud.storage.retry import (
     DEFAULT_RETRY,
     DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
@@ -64,6 +66,7 @@ def _get_default_timeout():
     def _make_client(*args, **kw):
         from google.cloud.storage.client import Client
 
+        kw["api_endpoint"] = kw.get("api_endpoint") or _get_default_storage_base_url()
         return mock.create_autospec(Client, instance=True, **kw)
 
     def test_ctor_wo_encryption_key(self):
@@ -426,6 +429,15 @@ def test_public_url_with_non_ascii(self):
         expected_url = "https://storage.googleapis.com/name/winter%20%E2%98%83"
         self.assertEqual(blob.public_url, expected_url)
 
+    def test_public_url_without_client(self):
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        bucket.client = None
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+        self.assertEqual(
+            blob.public_url, f"https://storage.googleapis.com/name/{BLOB_NAME}"
+        )
+
     def test_generate_signed_url_w_invalid_version(self):
         BLOB_NAME = "blob-name"
         EXPIRATION = "2014-10-16T20:34:37.000Z"
@@ -461,11 +473,9 @@ def _generate_signed_url_helper(
         from urllib import parse
         from google.cloud._helpers import UTC
         from google.cloud.storage._helpers import _bucket_bound_hostname_url
-        from google.cloud.storage.blob import _API_ACCESS_ENDPOINT
+        from google.cloud.storage._helpers import _get_default_storage_base_url
         from google.cloud.storage.blob import _get_encryption_headers
 
-        api_access_endpoint = api_access_endpoint or _API_ACCESS_ENDPOINT
-
         delta = datetime.timedelta(hours=1)
 
         if expiration is None:
@@ -522,7 +532,11 @@ def _generate_signed_url_helper(
                 bucket_bound_hostname, scheme
             )
         else:
-            expected_api_access_endpoint = api_access_endpoint
+            expected_api_access_endpoint = (
+                api_access_endpoint
+                if api_access_endpoint
+                else _get_default_storage_base_url()
+            )
             expected_resource = f"/{bucket.name}/{quoted_name}"
 
         if virtual_hosted_style or bucket_bound_hostname:
@@ -694,6 +708,17 @@ def test_generate_signed_url_v4_w_credentials(self):
         credentials = object()
         self._generate_signed_url_v4_helper(credentials=credentials)
 
+    def test_generate_signed_url_v4_w_incompatible_params(self):
+        with self.assertRaises(ValueError):
+            self._generate_signed_url_v4_helper(
+                api_access_endpoint="example.com",
+                bucket_bound_hostname="cdn.example.com",
+            )
+        with self.assertRaises(ValueError):
+            self._generate_signed_url_v4_helper(
+                virtual_hosted_style=True, bucket_bound_hostname="cdn.example.com"
+            )
+
     def test_exists_miss_w_defaults(self):
         from google.cloud.exceptions import NotFound
 
@@ -5905,7 +5930,10 @@ def test_downloads_w_client_custom_headers(self):
             "x-goog-custom-audit-foo": "bar",
             "x-goog-custom-audit-user": "baz",
         }
-        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+        credentials = mock.Mock(
+            spec=google.auth.credentials.Credentials,
+            universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+        )
         client = Client(
             project="project", credentials=credentials, extra_headers=custom_headers
         )
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 1b21e097a..642b09158 100644
--- a/tests/unit/test_bucket.py
+++ b/tests/unit/test_bucket.py
@@ -27,6 +27,7 @@
 from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
 from google.cloud.storage.constants import RPO_DEFAULT
 from google.cloud.storage.constants import RPO_ASYNC_TURBO
+from google.cloud.storage._helpers import _get_default_storage_base_url
 
 
 def _create_signing_credentials():
@@ -608,6 +609,7 @@ def _get_default_timeout():
     def _make_client(**kw):
         from google.cloud.storage.client import Client
 
+        kw["api_endpoint"] = kw.get("api_endpoint") or _get_default_storage_base_url()
         return mock.create_autospec(Client, instance=True, **kw)
 
     def _make_one(self, client=None, name=None, properties=None, user_project=None):
@@ -4058,9 +4060,7 @@ def _generate_signed_url_helper(
         from urllib import parse
         from google.cloud._helpers import UTC
         from google.cloud.storage._helpers import _bucket_bound_hostname_url
-        from google.cloud.storage.blob import _API_ACCESS_ENDPOINT
-
-        api_access_endpoint = api_access_endpoint or _API_ACCESS_ENDPOINT
+        from google.cloud.storage._helpers import _get_default_storage_base_url
 
         delta = datetime.timedelta(hours=1)
 
@@ -4108,7 +4108,9 @@ def _generate_signed_url_helper(
                 bucket_bound_hostname, scheme
             )
         else:
-            expected_api_access_endpoint = api_access_endpoint
+            expected_api_access_endpoint = (
+                api_access_endpoint or _get_default_storage_base_url()
+            )
             expected_resource = f"/{parse.quote(bucket_name)}"
 
         if virtual_hosted_style or bucket_bound_hostname:
@@ -4258,6 +4260,17 @@ def test_generate_signed_url_v4_w_bucket_bound_hostname_w_scheme(self):
     def test_generate_signed_url_v4_w_bucket_bound_hostname_w_bare_hostname(self):
         self._generate_signed_url_v4_helper(bucket_bound_hostname="cdn.example.com")
 
+    def test_generate_signed_url_v4_w_incompatible_params(self):
+        with self.assertRaises(ValueError):
+            self._generate_signed_url_v4_helper(
+                api_access_endpoint="example.com",
+                bucket_bound_hostname="cdn.example.com",
+            )
+        with self.assertRaises(ValueError):
+            self._generate_signed_url_v4_helper(
+                virtual_hosted_style=True, bucket_bound_hostname="cdn.example.com"
+            )
+
 
 class Test__item_to_notification(unittest.TestCase):
     def _call_fut(self, iterator, item):
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index 9650de976..8e1130227 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -30,7 +30,9 @@
 
 from google.cloud.storage import _helpers
 from google.cloud.storage._helpers import STORAGE_EMULATOR_ENV_VAR
+from google.cloud.storage._helpers import _API_ENDPOINT_OVERRIDE_ENV_VAR
 from google.cloud.storage._helpers import _get_default_headers
+from google.cloud.storage._helpers import _DEFAULT_UNIVERSE_DOMAIN
 from google.cloud.storage._http import Connection
 from google.cloud.storage.retry import DEFAULT_RETRY
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
@@ -45,13 +47,19 @@
 _FAKE_CREDENTIALS = Credentials.from_service_account_info(_SERVICE_ACCOUNT_JSON)
 
 
-def _make_credentials(project=None):
+def _make_credentials(project=None, universe_domain=_DEFAULT_UNIVERSE_DOMAIN):
     import google.auth.credentials
 
     if project is not None:
-        return mock.Mock(spec=google.auth.credentials.Credentials, project_id=project)
+        return mock.Mock(
+            spec=google.auth.credentials.Credentials,
+            project_id=project,
+            universe_domain=universe_domain,
+        )
 
-    return mock.Mock(spec=google.auth.credentials.Credentials)
+    return mock.Mock(
+        spec=google.auth.credentials.Credentials, universe_domain=universe_domain
+    )
 
 
 def _create_signing_credentials():
@@ -62,7 +70,9 @@ class _SigningCredentials(
     ):
         pass
 
-    credentials = mock.Mock(spec=_SigningCredentials)
+    credentials = mock.Mock(
+        spec=_SigningCredentials, universe_domain=_DEFAULT_UNIVERSE_DOMAIN
+    )
     credentials.sign_bytes = mock.Mock(return_value=b"Signature_bytes")
     credentials.signer_email = "test@mail.com"
     return credentials
@@ -162,22 +172,63 @@ def test_ctor_w_client_options_dict(self):
         )
 
         self.assertEqual(client._connection.API_BASE_URL, api_endpoint)
+        self.assertEqual(client.api_endpoint, api_endpoint)
 
     def test_ctor_w_client_options_object(self):
         from google.api_core.client_options import ClientOptions
 
         PROJECT = "PROJECT"
         credentials = _make_credentials()
-        client_options = ClientOptions(api_endpoint="https://www.foo-googleapis.com")
+        api_endpoint = "https://www.foo-googleapis.com"
+        client_options = ClientOptions(api_endpoint=api_endpoint)
 
         client = self._make_one(
             project=PROJECT, credentials=credentials, client_options=client_options
         )
 
-        self.assertEqual(
-            client._connection.API_BASE_URL, "https://www.foo-googleapis.com"
+        self.assertEqual(client._connection.API_BASE_URL, api_endpoint)
+        self.assertEqual(client.api_endpoint, api_endpoint)
+
+    def test_ctor_w_universe_domain_and_matched_credentials(self):
+        PROJECT = "PROJECT"
+        universe_domain = "example.com"
+        expected_api_endpoint = f"https://storage.{universe_domain}"
+        credentials = _make_credentials(universe_domain=universe_domain)
+        client_options = {"universe_domain": universe_domain}
+
+        client = self._make_one(
+            project=PROJECT, credentials=credentials, client_options=client_options
         )
 
+        self.assertEqual(client._connection.API_BASE_URL, expected_api_endpoint)
+        self.assertEqual(client.api_endpoint, expected_api_endpoint)
+        self.assertEqual(client.universe_domain, universe_domain)
+
+    def test_ctor_w_universe_domain_and_mismatched_credentials(self):
+        PROJECT = "PROJECT"
+        universe_domain = "example.com"
+        credentials = _make_credentials()  # default universe domain
+        client_options = {"universe_domain": universe_domain}
+
+        with self.assertRaises(ValueError):
+            self._make_one(
+                project=PROJECT, credentials=credentials, client_options=client_options
+            )
+
+    def test_ctor_w_universe_domain_and_mtls(self):
+        PROJECT = "PROJECT"
+        universe_domain = "example.com"
+        client_options = {"universe_domain": universe_domain}
+
+        credentials = _make_credentials(
+            project=PROJECT, universe_domain=universe_domain
+        )
+
+        environ = {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        with mock.patch("os.environ", environ):
+            with self.assertRaises(ValueError):
+                self._make_one(credentials=credentials, client_options=client_options)
+
     def test_ctor_w_custom_headers(self):
         PROJECT = "PROJECT"
         credentials = _make_credentials()
@@ -330,6 +381,16 @@ def test_ctor_w_emulator_w_credentials(self):
         self.assertEqual(client._connection.API_BASE_URL, host)
         self.assertIs(client._connection.credentials, credentials)
 
+    def test_ctor_w_api_endpoint_override(self):
+        host = "http://localhost:8080"
+        environ = {_API_ENDPOINT_OVERRIDE_ENV_VAR: host}
+        project = "my-test-project"
+        with mock.patch("os.environ", environ):
+            client = self._make_one(project=project)
+
+        self.assertEqual(client.project, project)
+        self.assertEqual(client._connection.API_BASE_URL, host)
+
     def test_create_anonymous_client(self):
         klass = self._get_target_class()
         client = klass.create_anonymous_client()
@@ -2677,6 +2738,25 @@ def test_get_signed_policy_v4_bucket_bound_hostname(self):
             )
         self.assertEqual(policy["url"], "https://bucket.bound_hostname/")
 
+    def test_get_signed_policy_v4_with_conflicting_arguments(self):
+        import datetime
+
+        project = "PROJECT"
+        credentials = _make_credentials(project=project)
+        client = self._make_one(credentials=credentials)
+
+        dtstamps_patch, _, _ = _time_functions_patches()
+        with dtstamps_patch:
+            with self.assertRaises(ValueError):
+                client.generate_signed_post_policy_v4(
+                    "bucket-name",
+                    "object-name",
+                    expiration=datetime.datetime(2020, 3, 12),
+                    bucket_bound_hostname="https://bucket.bound_hostname",
+                    virtual_hosted_style=True,
+                    credentials=_create_signing_credentials(),
+                )
+
     def test_get_signed_policy_v4_bucket_bound_hostname_with_scheme(self):
         import datetime