chore: a refactoring of DEFAULT_UNIVERSE_DOMAIN which is widely used in code and tests

google/auth/external_account.py

@@ -51,8 +51,6 @@
 # Cloud resource manager URL used to retrieve project information.
 _CLOUD_RESOURCE_MANAGER = "https://cloudresourcemanager.googleapis.com/v1/projects/"
 
-_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
-
 
 class Credentials(
     credentials.Scoped,
@@ -83,7 +81,7 @@ def __init__(
         scopes=None,
         default_scopes=None,
         workforce_pool_user_project=None,
-        universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
         trust_boundary=None,
     ):
         """Instantiates an external account credentials object.
@@ -131,7 +129,7 @@ def __init__(
         self._scopes = scopes
         self._default_scopes = default_scopes
         self._workforce_pool_user_project = workforce_pool_user_project
-        self._universe_domain = universe_domain or _DEFAULT_UNIVERSE_DOMAIN
+        self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
         self._trust_boundary = {
             "locations": [],
             "encoded_locations": "0x0",
@@ -513,7 +511,9 @@ def from_info(cls, info, **kwargs):
             credential_source=info.get("credential_source"),
             quota_project_id=info.get("quota_project_id"),
             workforce_pool_user_project=info.get("workforce_pool_user_project"),
-            universe_domain=info.get("universe_domain", _DEFAULT_UNIVERSE_DOMAIN),
+            universe_domain=info.get(
+                "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
+            ),
             **kwargs
         )
 

google/auth/external_account_authorized_user.py

@@ -43,7 +43,6 @@
 from google.oauth2 import sts
 from google.oauth2 import utils
 
-_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
 _EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE = "external_account_authorized_user"
 
 
@@ -76,7 +75,7 @@ def __init__(
         revoke_url=None,
         scopes=None,
         quota_project_id=None,
-        universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
     ):
         """Instantiates a external account authorized user credentials object.
 
@@ -120,7 +119,7 @@ def __init__(
         self._revoke_url = revoke_url
         self._quota_project_id = quota_project_id
         self._scopes = scopes
-        self._universe_domain = universe_domain or _DEFAULT_UNIVERSE_DOMAIN
+        self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
 
         if not self.valid and not self.can_refresh:
             raise exceptions.InvalidOperation(
@@ -342,7 +341,9 @@ def from_info(cls, info, **kwargs):
             revoke_url=info.get("revoke_url"),
             quota_project_id=info.get("quota_project_id"),
             scopes=info.get("scopes"),
-            universe_domain=info.get("universe_domain", _DEFAULT_UNIVERSE_DOMAIN),
+            universe_domain=info.get(
+                "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
+            ),
             **kwargs
         )
 

google/oauth2/credentials.py

@@ -49,7 +49,6 @@
 
 # The Google OAuth 2.0 token endpoint. Used for authorized user credentials.
 _GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
-_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
 
 
 class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaProject):
@@ -86,7 +85,7 @@ def __init__(
         enable_reauth_refresh=False,
         granted_scopes=None,
         trust_boundary=None,
-        universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
         account=None,
     ):
         """
@@ -150,7 +149,7 @@ def __init__(
         self.refresh_handler = refresh_handler
         self._enable_reauth_refresh = enable_reauth_refresh
         self._trust_boundary = trust_boundary
-        self._universe_domain = universe_domain or _DEFAULT_UNIVERSE_DOMAIN
+        self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
         self._account = account or ""
 
     def __getstate__(self):
@@ -187,7 +186,9 @@ def __setstate__(self, d):
         self._rapt_token = d.get("_rapt_token")
         self._enable_reauth_refresh = d.get("_enable_reauth_refresh")
         self._trust_boundary = d.get("_trust_boundary")
-        self._universe_domain = d.get("_universe_domain") or _DEFAULT_UNIVERSE_DOMAIN
+        self._universe_domain = (
+            d.get("_universe_domain") or credentials.DEFAULT_UNIVERSE_DOMAIN
+        )
         # The refresh_handler setter should be used to repopulate this.
         self._refresh_handler = None
         self._refresh_worker = None
@@ -373,7 +374,7 @@ def _metric_header_for_usage(self):
 
     @_helpers.copy_docstring(credentials.Credentials)
     def refresh(self, request):
-        if self._universe_domain != _DEFAULT_UNIVERSE_DOMAIN:
+        if self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
             raise exceptions.RefreshError(
                 "User credential refresh is only supported in the default "
                 "googleapis.com universe domain, but the current universe "

google/oauth2/service_account.py

@@ -82,7 +82,6 @@
 from google.oauth2 import _client
 
 _DEFAULT_TOKEN_LIFETIME_SECS = 3600  # 1 hour in seconds
-_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
 _GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
 
 
@@ -139,7 +138,7 @@ def __init__(
         quota_project_id=None,
         additional_claims=None,
         always_use_jwt_access=False,
-        universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
         trust_boundary=None,
     ):
         """
@@ -182,9 +181,9 @@ def __init__(
         self._quota_project_id = quota_project_id
         self._token_uri = token_uri
         self._always_use_jwt_access = always_use_jwt_access
-        self._universe_domain = universe_domain or _DEFAULT_UNIVERSE_DOMAIN
+        self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
 
-        if universe_domain != _DEFAULT_UNIVERSE_DOMAIN:
+        if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
             self._always_use_jwt_access = True
 
         self._jwt_credentials = None
@@ -216,7 +215,9 @@ def _from_signer_and_info(cls, signer, info, **kwargs):
             service_account_email=info["client_email"],
             token_uri=info["token_uri"],
             project_id=info.get("project_id"),
-            universe_domain=info.get("universe_domain", _DEFAULT_UNIVERSE_DOMAIN),
+            universe_domain=info.get(
+                "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
+            ),
             trust_boundary=info.get("trust_boundary"),
             **kwargs
         )
@@ -316,7 +317,7 @@ def with_always_use_jwt_access(self, always_use_jwt_access):
         """
         cred = self._make_copy()
         if (
-            cred._universe_domain != _DEFAULT_UNIVERSE_DOMAIN
+            cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
             and not always_use_jwt_access
         ):
             raise exceptions.InvalidValue(
@@ -329,7 +330,7 @@ def with_always_use_jwt_access(self, always_use_jwt_access):
     def with_universe_domain(self, universe_domain):
         cred = self._make_copy()
         cred._universe_domain = universe_domain
-        if universe_domain != _DEFAULT_UNIVERSE_DOMAIN:
+        if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
             cred._always_use_jwt_access = True
         return cred
 
@@ -427,7 +428,10 @@ def refresh(self, request):
             # created, try to create one with scopes
             self._create_self_signed_jwt(None)
 
-        if self._universe_domain != _DEFAULT_UNIVERSE_DOMAIN and self._subject:
+        if (
+            self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
+            and self._subject
+        ):
             raise exceptions.RefreshError(
                 "domain wide delegation is not supported for non-default universe domain"
             )
@@ -556,7 +560,7 @@ def __init__(
         target_audience,
         additional_claims=None,
         quota_project_id=None,
-        universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
     ):
         """
         Args:
@@ -588,11 +592,11 @@ def __init__(
         self._use_iam_endpoint = False
 
         if not universe_domain:
-            self._universe_domain = _DEFAULT_UNIVERSE_DOMAIN
+            self._universe_domain = credentials.DEFAULT_UNIVERSE_DOMAIN
         else:
             self._universe_domain = universe_domain
 
-        if universe_domain != _DEFAULT_UNIVERSE_DOMAIN:
+        if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
             self._use_iam_endpoint = True
 
         if additional_claims is not None:
@@ -708,7 +712,10 @@ def _with_use_iam_endpoint(self, use_iam_endpoint):
                 default and use_iam_endpoint is False.
         """
         cred = self._make_copy()
-        if cred._universe_domain != _DEFAULT_UNIVERSE_DOMAIN and not use_iam_endpoint:
+        if (
+            cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
+            and not use_iam_endpoint
+        ):
             raise exceptions.InvalidValue(
                 "use_iam_endpoint should be True for non-default universe domain"
             )

tests/oauth2/test_service_account.py

@@ -24,6 +24,7 @@
 from google.auth import exceptions
 from google.auth import jwt
 from google.auth import transport
+from google.auth.credentials import DEFAULT_UNIVERSE_DOMAIN
 from google.oauth2 import service_account
 
 
@@ -58,7 +59,7 @@ class TestCredentials(object):
     TOKEN_URI = "https://example.com/oauth2/token"
 
     @classmethod
-    def make_credentials(cls, universe_domain=service_account._DEFAULT_UNIVERSE_DOMAIN):
+    def make_credentials(cls, universe_domain=DEFAULT_UNIVERSE_DOMAIN):
         return service_account.Credentials(
             SIGNER,
             cls.SERVICE_ACCOUNT_EMAIL,
@@ -70,7 +71,7 @@ def test_constructor_no_universe_domain(self):
         credentials = service_account.Credentials(
             SIGNER, self.SERVICE_ACCOUNT_EMAIL, self.TOKEN_URI, universe_domain=None
         )
-        assert credentials.universe_domain == service_account._DEFAULT_UNIVERSE_DOMAIN
+        assert credentials.universe_domain == DEFAULT_UNIVERSE_DOMAIN
 
     def test_from_service_account_info(self):
         credentials = service_account.Credentials.from_service_account_info(
@@ -80,7 +81,7 @@ def test_from_service_account_info(self):
         assert credentials._signer.key_id == SERVICE_ACCOUNT_INFO["private_key_id"]
         assert credentials.service_account_email == SERVICE_ACCOUNT_INFO["client_email"]
         assert credentials._token_uri == SERVICE_ACCOUNT_INFO["token_uri"]
-        assert credentials._universe_domain == service_account._DEFAULT_UNIVERSE_DOMAIN
+        assert credentials._universe_domain == DEFAULT_UNIVERSE_DOMAIN
         assert not credentials._always_use_jwt_access
 
     def test_from_service_account_info_non_gdu(self):
@@ -595,7 +596,7 @@ class TestIDTokenCredentials(object):
     TARGET_AUDIENCE = "https://example.com"
 
     @classmethod
-    def make_credentials(cls, universe_domain=service_account._DEFAULT_UNIVERSE_DOMAIN):
+    def make_credentials(cls, universe_domain=DEFAULT_UNIVERSE_DOMAIN):
         return service_account.IDTokenCredentials(
             SIGNER,
             cls.SERVICE_ACCOUNT_EMAIL,
@@ -612,7 +613,7 @@ def test_constructor_no_universe_domain(self):
             self.TARGET_AUDIENCE,
             universe_domain=None,
         )
-        assert credentials._universe_domain == service_account._DEFAULT_UNIVERSE_DOMAIN
+        assert credentials._universe_domain == DEFAULT_UNIVERSE_DOMAIN
 
     def test_from_service_account_info(self):
         credentials = service_account.IDTokenCredentials.from_service_account_info(

tests/test_aws.py

@@ -26,7 +26,7 @@
 from google.auth import environment_vars
 from google.auth import exceptions
 from google.auth import transport
-
+from google.auth.credentials import DEFAULT_UNIVERSE_DOMAIN
 
 IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE = (
     "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/imp"
@@ -75,7 +75,6 @@
 # Each tuple contains the following entries:
 # region, time, credentials, original_request, signed_request
 
-DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
 VALID_TOKEN_URLS = [
     "https://sts.googleapis.com",
     "https://us-east-1.sts.googleapis.com",

tests/test_external_account.py

@@ -24,9 +24,9 @@
 from google.auth import exceptions
 from google.auth import external_account
 from google.auth import transport
+from google.auth.credentials import DEFAULT_UNIVERSE_DOMAIN
 from google.auth.credentials import TokenState
 
-
 IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE = (
     "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/imp"
 )
@@ -150,7 +150,7 @@ def make_credentials(
         default_scopes=None,
         service_account_impersonation_url=None,
         service_account_impersonation_options={},
-        universe_domain=external_account._DEFAULT_UNIVERSE_DOMAIN,
+        universe_domain=DEFAULT_UNIVERSE_DOMAIN,
     ):
         return CredentialsImpl(
             audience=cls.AUDIENCE,
@@ -386,7 +386,7 @@ def test_with_scopes_full_options_propagated(self):
             quota_project_id=self.QUOTA_PROJECT_ID,
             scopes=["email"],
             default_scopes=["default2"],
-            universe_domain=external_account._DEFAULT_UNIVERSE_DOMAIN,
+            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
         )
 
     def test_with_token_uri(self):
@@ -474,7 +474,7 @@ def test_with_quota_project_full_options_propagated(self):
             quota_project_id="project-foo",
             scopes=self.SCOPES,
             default_scopes=["default1"],
-            universe_domain=external_account._DEFAULT_UNIVERSE_DOMAIN,
+            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
         )
 
     def test_with_invalid_impersonation_target_principal(self):
@@ -504,7 +504,7 @@ def test_universe_domain(self):
         assert credentials.universe_domain == "dummy_universe.com"
 
         credentials = self.make_credentials()
-        assert credentials.universe_domain == external_account._DEFAULT_UNIVERSE_DOMAIN
+        assert credentials.universe_domain == DEFAULT_UNIVERSE_DOMAIN
 
     def test_with_universe_domain(self):
         credentials = self.make_credentials()
@@ -523,7 +523,7 @@ def test_info_workforce_pool(self):
             "token_url": self.TOKEN_URL,
             "credential_source": self.CREDENTIAL_SOURCE.copy(),
             "workforce_pool_user_project": self.WORKFORCE_POOL_USER_PROJECT,
-            "universe_domain": external_account._DEFAULT_UNIVERSE_DOMAIN,
+            "universe_domain": DEFAULT_UNIVERSE_DOMAIN,
         }
 
     def test_info_with_full_options(self):
@@ -548,7 +548,7 @@ def test_info_with_full_options(self):
             "quota_project_id": self.QUOTA_PROJECT_ID,
             "client_id": CLIENT_ID,
             "client_secret": CLIENT_SECRET,
-            "universe_domain": external_account._DEFAULT_UNIVERSE_DOMAIN,
+            "universe_domain": DEFAULT_UNIVERSE_DOMAIN,
         }
 
     def test_service_account_email_without_impersonation(self):

tests/test_external_account_authorized_user.py

@@ -22,6 +22,7 @@
 from google.auth import exceptions
 from google.auth import external_account_authorized_user
 from google.auth import transport
+from google.auth.credentials import DEFAULT_UNIVERSE_DOMAIN
 
 TOKEN_URL = "https://sts.googleapis.com/v1/token"
 TOKEN_INFO_URL = "https://sts.googleapis.com/v1/introspect"
@@ -45,7 +46,6 @@
 SCOPES = ["email", "profile"]
 NOW = datetime.datetime(1990, 8, 27, 6, 54, 30)
 FAKE_UNIVERSE_DOMAIN = "fake-universe-domain"
-DEFAULT_UNIVERSE_DOMAIN = external_account_authorized_user._DEFAULT_UNIVERSE_DOMAIN
 
 
 class TestCredentials(object):

tests/test_identity_pool.py

@@ -26,7 +26,7 @@
 from google.auth import identity_pool
 from google.auth import metrics
 from google.auth import transport
-
+from google.auth.credentials import DEFAULT_UNIVERSE_DOMAIN
 
 CLIENT_ID = "username"
 CLIENT_SECRET = "password"
@@ -67,8 +67,6 @@
 WORKFORCE_SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:id_token"
 WORKFORCE_POOL_USER_PROJECT = "WORKFORCE_POOL_USER_PROJECT_NUMBER"
 
-DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
-
 VALID_TOKEN_URLS = [
     "https://sts.googleapis.com",
     "https://us-east-1.sts.googleapis.com",