From 3d93dc40a2c1699670776b699955e9f905c0b15a Mon Sep 17 00:00:00 2001 From: baha-zrelli Date: Thu, 2 Apr 2026 23:51:19 +0200 Subject: [PATCH] feat(google-auth): make _CLOUD_RESOURCE_MANAGER URL universe-domain-aware Replace hardcoded googleapis.com in _CLOUD_RESOURCE_MANAGER with a {universe_domain} placeholder, resolved at credential construction time via self._cloud_resource_manager_url. This mirrors the existing pattern used for _DEFAULT_TOKEN_URL. Add tests verifying the URL is correctly built for both the default (googleapis.com) and custom universe domains, including an end-to-end test through get_project_id. --- .../google/auth/external_account.py | 7 ++-- .../tests/test_external_account.py | 36 +++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/packages/google-auth/google/auth/external_account.py b/packages/google-auth/google/auth/external_account.py index 05874eda7fb1..9bd8b9fe479a 100644 --- a/packages/google-auth/google/auth/external_account.py +++ b/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) diff --git a/packages/google-auth/tests/test_external_account.py b/packages/google-auth/tests/test_external_account.py index e50a1b8c0c5c..bf54c3e5e3ab 100644 --- a/packages/google-auth/tests/test_external_account.py +++ b/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