Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions packages/google-auth/google/auth/external_account.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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
)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A couple inconsequential comments:

  • do we actually need to save this as an attribute? Seems like it's still only used in one place
  • since we're using {universe_domain} as a template, maybe we should handle the project_number in the same way, instead of concatenating it on the end

Copy link
Copy Markdown
Author

@baha-zrelli baha-zrelli Apr 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Tried to follow similar behaviour of previous PR so I've limited myself on made changes, would be good if we can go with this to unblock partner and do those changes later on.

self._token_info_url = token_info_url
self._credential_source = credential_source
self._service_account_impersonation_url = service_account_impersonation_url
Expand Down Expand Up @@ -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
Comment thread
baha-zrelli marked this conversation as resolved.
Outdated
self.before_request(request, "GET", url, headers)
response = request(url=url, method="GET", headers=headers)

Expand Down
36 changes: 36 additions & 0 deletions packages/google-auth/tests/test_external_account.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading