feat(google-auth): make _CLOUD_RESOURCE_MANAGER URL universe-domain-aware

packages/google-auth/google/auth/external_account.py

@@ -52,7 +52,7 @@
 # The token exchange requested_token_type. This is always an access_token.
 _STS_REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
 # Cloud resource manager URL used to retrieve project information.
-_CLOUD_RESOURCE_MANAGER = "https://cloudresourcemanager.googleapis.com/v1/projects/"
+_CLOUD_RESOURCE_MANAGER = "https://cloudresourcemanager.{universe_domain}/v1/projects/"
 # Default Google sts token url.
 _DEFAULT_TOKEN_URL = "https://sts.{universe_domain}/v1/token"
 
@@ -164,6 +164,9 @@ def __init__(
             self._token_url = self._token_url.replace(
                 "{universe_domain}", self._universe_domain
             )
+        self._cloud_resource_manager_url = _CLOUD_RESOURCE_MANAGER.replace(
+            "{universe_domain}", self._universe_domain
+        )
         self._token_info_url = token_info_url
         self._credential_source = credential_source
         self._service_account_impersonation_url = service_account_impersonation_url
@@ -395,7 +398,7 @@ def get_project_id(self, request):
         project_number = self.project_number or self._workforce_pool_user_project
         if project_number and scopes:
             headers = {}
-            url = _CLOUD_RESOURCE_MANAGER + project_number
+            url = self._cloud_resource_manager_url + project_number
             self.before_request(request, "GET", url, headers)
             response = request(url=url, method="GET", headers=headers)
 

packages/google-auth/tests/test_external_account.py

@@ -2404,6 +2404,42 @@ def test_get_project_id_cloud_resource_manager_error(self):
         # Only 2 requests to STS and cloud resource manager should be sent.
         assert len(request.call_args_list) == 2
 
+    def test_cloud_resource_manager_url_with_default_universe_domain(self):
+        credentials = self.make_credentials()
+        assert credentials._cloud_resource_manager_url == (
+            "https://cloudresourcemanager.googleapis.com/v1/projects/"
+        )
+
+    def test_cloud_resource_manager_url_with_custom_universe_domain(self):
+        credentials = self.make_credentials(universe_domain="example.com")
+        assert credentials._cloud_resource_manager_url == (
+            "https://cloudresourcemanager.example.com/v1/projects/"
+        )
+
+    def test_get_project_id_cloud_resource_manager_custom_universe_domain(self):
+        custom_universe_domain = "example.com"
+        request = self.make_mock_request(
+            status=http_client.OK,
+            data=self.SUCCESS_RESPONSE.copy(),
+            cloud_resource_manager_status=http_client.OK,
+            cloud_resource_manager_data=self.CLOUD_RESOURCE_MANAGER_SUCCESS_RESPONSE,
+        )
+        credentials = self.make_credentials(
+            scopes=self.SCOPES,
+            universe_domain=custom_universe_domain,
+        )
+
+        project_id = credentials.get_project_id(request)
+
+        assert project_id == self.PROJECT_ID
+        # Verify that the cloud resource manager request used the custom universe domain URL.
+        assert len(request.call_args_list) == 2
+        crm_request_kwargs = request.call_args_list[1][1]
+        expected_url = "https://cloudresourcemanager.{}/v1/projects/{}".format(
+            custom_universe_domain, self.PROJECT_NUMBER
+        )
+        assert crm_request_kwargs["url"] == expected_url
+
     def test_refresh_with_existing_impersonated_credentials(self):
         credentials = self.make_credentials(
             service_account_impersonation_url=self.SERVICE_ACCOUNT_IMPERSONATION_URL