feat: add support to validate credentials for configured universe

googleapiclient/discovery.py

@@ -53,6 +53,13 @@
 except ImportError:  # pragma: NO COVER
     google_auth_httplib2 = None
 
+try:
+    from google.api_core import universe
+
+    HAS_UNIVERSE = True
+except ImportError:
+    HAS_UNIVERSE = False
+
 # Local imports
 from googleapiclient import _auth, mimeparse
 from googleapiclient._helpers import _add_query_parameter, positional
@@ -1352,6 +1359,7 @@ def __init__(
         resourceDesc,
         rootDesc,
         schema,
+        universe_domain=universe.DEFAULT_UNIVERSE if HAS_UNIVERSE else "",
     ):
         """Build a Resource from the API description.
 
@@ -1369,6 +1377,8 @@ def __init__(
               is considered a resource.
           rootDesc: object, the entire deserialized discovery document.
           schema: object, mapping of schema names to schema descriptions.
+          universe_domain: string, the universe for the API. The default universe
+          is "googleapis.com".
         """
         self._dynamic_attrs = []
 
@@ -1380,6 +1390,8 @@ def __init__(
         self._resourceDesc = resourceDesc
         self._rootDesc = rootDesc
         self._schema = schema
+        self._universe_domain = universe_domain
+        self._credentials_validated = False
 
         self._set_service_methods()
 
@@ -1546,6 +1558,27 @@ def _add_next_methods(self, resourceDesc, schema):
                 fixedMethodName, method.__get__(self, self.__class__)
             )
 
+    def _validate_credentials(self):
+        """Validates client's and credentials' universe domains are consistent.
+
+        Returns:
+            bool: True iff the configured universe domain is valid.
+
+        Raises:
+            UniverseMismatchError: If the configured universe domain is not valid.
+        """
+        credentials = getattr(self._http, "credentials", None)
+
+        self._credentials_validated = (
+            (
+                self._credentials_validated
+                or universe.compare_domains(self._universe_domain, credentials)
+            )
+            if HAS_UNIVERSE
+            else True
+        )
+        return self._credentials_validated
+
 
 def _findPageTokenName(fields):
     """Search field names for one like a page token.

tests/test_discovery.py

@@ -54,6 +54,13 @@
 except ImportError:
     HAS_OAUTH2CLIENT = False
 
+try:
+    from google.api_core import universe
+
+    HAS_UNIVERSE = True
+except ImportError:
+    HAS_UNIVERSE = False
+
 from googleapiclient import _helpers as util
 from googleapiclient.discovery import (
     DISCOVERY_URI,
@@ -2118,6 +2125,7 @@ def test_resumable_media_handle_resume_of_upload_of_unknown_size(self):
     def test_pickle(self):
         sorted_resource_keys = [
             "_baseUrl",
+            "_credentials_validated",
             "_developerKey",
             "_dynamic_attrs",
             "_http",
@@ -2126,6 +2134,7 @@ def test_pickle(self):
             "_resourceDesc",
             "_rootDesc",
             "_schema",
+            "_universe_domain",
             "animals",
             "global_",
             "load",
@@ -2331,5 +2340,94 @@ def test_get_media(self):
         self.assertEqual(b"standing in for media", response)
 
 
+if HAS_UNIVERSE:
+
+    class Universe(unittest.TestCase):
+        def test_validate_credentials_with_no_universe(self):
+            fake_universe = "foo.com"
+
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=None, http=build_http()
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=universe.DEFAULT_UNIVERSE
+                ),
+            )
+
+            assert service._validate_credentials()
+
+            # TODO(google-api-python-client/issues/2365): Add test case for no configured  universe and fake credentials' universe.
+
+            # TODO(google-api-python-client/issues/2365): Add test case for not specifying universe domain via client option.
+
+        def test_validate_credentials_with_default_universe(self):
+            fake_universe = "foo.com"
+
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=mock.Mock(universe_domain=universe.DEFAULT_UNIVERSE),
+                http=build_http(),
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=universe.DEFAULT_UNIVERSE
+                ),
+            )
+
+            assert service._validate_credentials()
+
+            # TODO(google-api-python-client/issues/2365): # Add test case for "googleapis.com" configured universe and fake credentials' universe.
+
+        def test_validate_credentials_with_a_different_universe(self):
+            fake_universe = "foo.com"
+
+            # TODO(google-api-python-client/issues/2365): Add test case for fake configured universe and fake credentials' universe.
+
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=mock.Mock(universe_domain=fake_universe), http=build_http()
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=universe.DEFAULT_UNIVERSE
+                ),
+            )
+
+            with self.assertRaises(universe.UniverseMismatchError):
+                service._validate_credentials()
+
+        def test_validate_credentials_with_already_validated_credentials(self):
+            fake_universe = "foo.com"
+
+            http = google_auth_httplib2.AuthorizedHttp(
+                credentials=mock.Mock(universe_domain=universe.DEFAULT_UNIVERSE),
+                http=build_http(),
+            )
+            discovery = read_datafile("zoo.json")
+            service = build_from_document(
+                discovery,
+                http=http,
+                client_options=google.api_core.client_options.ClientOptions(
+                    universe_domain=universe.DEFAULT_UNIVERSE
+                ),
+            )
+
+            assert service._validate_credentials()
+            assert service._credentials_validated
+
+            # Calling service._validate_credentials() again returns service.credentials_validated.
+            assert service._validate_credentials()
+
+            # TODO(google-api-python-client/issues/2365): Add test case for fake configured universe and fake credentials' universe.
+
+
 if __name__ == "__main__":
     unittest.main()