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 as 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 None,
     ):
         """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 as 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,129 @@ 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(omairn): uncomment when universe configured universe is leveraged.
+            # 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=fake_universe))
+
+            # with self.assertRaises(universe.UniverseMismatchError):
+            #     service._validate_credentials()
+
+            # TODO(omairn): 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(omairn): uncomment when universe configured universe is leveraged.
+            # 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=fake_universe))
+
+            # with self.assertRaises(universe.UniverseMismatchError):
+            #     service._validate_credentials()
+
+        def test_validate_credentials_with_a_different_universe(self):
+            fake_universe = "foo.com"
+
+            # TODO(omairn): uncomment when universe configured universe is leveraged.
+            # 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=fake_universe))
+            # assert service._validate_credentials()
+
+            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
+            assert service._validate_credentials()
+
+            # TODO(omairn): uncomment when universe configured universe is leveraged.
+            # 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=fake_universe))
+
+            # assert service._validate_credentials()
+            # assert service._credentials_validated
+            # assert service._validate_credentials()
+
+        # TODO(omairn):
+        # Once _validate_universe() is added to the API call, verify that:
+        # the API call succeeds when the credentials match.
+        # the API call fails when there is a mismatch.
+        # the credentials are not verified twice for an already verified Resource.
+
+        # Add test case once universe domain is supported via environment variable.
+
+        # Once universe.determine_domain() is leveraged:
+        # add test case for empty universe.
+        # add test cases when universe domain is provided via both client option and env var.
+
+        # Add test case when credentials are passed instead of http.
+
+
 if __name__ == "__main__":
     unittest.main()