feat: implement get_project_id() for google.auth.external_account.Credentials

google/auth/compute_engine/_metadata.py

@@ -108,7 +108,9 @@ def ping(request, timeout=_METADATA_DEFAULT_TIMEOUT, retry_count=3):
     return False
 
 
-def get(request, path, root=_METADATA_ROOT, recursive=False, retry_count=5):
+def get(
+    request, path, root=_METADATA_ROOT, params=None, recursive=False, retry_count=5
+):
     """Fetch a resource from the metadata server.
 
     Args:
@@ -117,6 +119,8 @@ def get(request, path, root=_METADATA_ROOT, recursive=False, retry_count=5):
         path (str): The resource to retrieve. For example,
             ``'instance/service-accounts/default'``.
         root (str): The full path to the metadata server root.
+        params (Optional[Mapping[str, str]]): A mapping of query parameter
+            keys to values.
         recursive (bool): Whether to do a recursive query of metadata. See
             https://cloud.google.com/compute/docs/metadata#aggcontents for more
             details.
@@ -133,7 +137,7 @@ def get(request, path, root=_METADATA_ROOT, recursive=False, retry_count=5):
             retrieving metadata.
     """
     base_url = urlparse.urljoin(root, path)
-    query_params = {}
+    query_params = {} if params is None else params
 
     if recursive:
         query_params["recursive"] = "true"
@@ -224,11 +228,10 @@ def get_service_account_info(request, service_account="default"):
         google.auth.exceptions.TransportError: if an error occurred while
             retrieving metadata.
     """
-    return get(
-        request,
-        "instance/service-accounts/{0}/".format(service_account),
-        recursive=True,
-    )
+    path = "instance/service-accounts/{0}/".format(service_account)
+    # See https://cloud.google.com/compute/docs/metadata#aggcontents
+    # for more on the use of 'recursive'.
+    return get(request, path, params={"recursive": "true"})
 
 
 def get_service_account_token(request, service_account="default"):

google/auth/compute_engine/credentials.py

@@ -323,12 +323,9 @@ def _call_metadata_identity_endpoint(self, request):
             ValueError: If extracting expiry from the obtained ID token fails.
         """
         try:
-            id_token = _metadata.get(
-                request,
-                "instance/service-accounts/default/identity?audience={}&format=full".format(
-                    self._target_audience
-                ),
-            )
+            path = "instance/service-accounts/default/identity"
+            params = {"audience": self._target_audience, "format": "full"}
+            id_token = _metadata.get(request, path, params=params)
         except exceptions.TransportError as caught_exc:
             new_exc = exceptions.RefreshError(caught_exc)
             six.raise_from(new_exc, caught_exc)

google/auth/external_account.py

@@ -28,6 +28,7 @@
 
 import abc
 import datetime
+import json
 
 import six
 
@@ -42,6 +43,8 @@
 _STS_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
 # 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/"
 
 
 @six.add_metaclass(abc.ABCMeta)
@@ -107,6 +110,7 @@ def __init__(
             self._impersonated_credentials = self._initialize_impersonated_credentials()
         else:
             self._impersonated_credentials = None
+        self._project_id = None
 
     @property
     def requires_scopes(self):
@@ -117,6 +121,20 @@ def requires_scopes(self):
         """
         return True if not self._scopes else False
 
+    @property
+    def project_number(self):
+        """Optional[str]: The project number corresponding to the workload identity pool."""
+
+        # STS audience pattern:
+        # //iam.googleapis.com/projects/$PROJECT_NUMBER/locations/...
+        components = self._audience.split("/")
+        try:
+            project_index = components.index("projects")
+            if project_index + 1 < len(components):
+                return components[project_index + 1] or None
+        except ValueError:
+            return None
+
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes):
         return self.__class__(
@@ -144,6 +162,49 @@ def retrieve_subject_token(self, request):
         # (pylint doesn't recognize that this is abstract)
         raise NotImplementedError("retrieve_subject_token must be implemented")
 
+    def get_project_id(self, request):
+        """Retrieves the project ID corresponding to the workload identity pool.
+
+        When not determinable, None is returned.
+
+        This is introduced to support the current pattern of using the Auth library:
+
+            credentials, project_id = google.auth.default()
+
+        The resource may not have permission (resourcemanager.projects.get) to
+        call this API or the required scopes may not be selected:
+        https://cloud.google.com/resource-manager/reference/rest/v1/projects/get#authorization-scopes
+
+        Args:
+            request (google.auth.transport.Request): A callable used to make
+                HTTP requests.
+        Returns:
+            Optional[str]: The project ID corresponding to the workload identity pool
+                if determinable.
+        """
+        if self._project_id:
+            # If already retrieved, return the cached project ID value.
+            return self._project_id
+        if self.project_number:
+            headers = {}
+            url = _CLOUD_RESOURCE_MANAGER + self.project_number
+            self.before_request(request, "GET", url, headers)
+            response = request(url=url, method="GET", headers=headers)
+
+            response_body = (
+                response.data.decode("utf-8")
+                if hasattr(response.data, "decode")
+                else response.data
+            )
+            response_data = json.loads(response_body)
+
+            if response.status == 200:
+                # Cache result as this field is immutable.
+                self._project_id = response_data.get("projectId")
+                return self._project_id
+
+        return None
+
     @_helpers.copy_docstring(credentials.Credentials)
     def refresh(self, request):
         if self._impersonated_credentials:

noxfile.py

@@ -142,6 +142,7 @@ def docs(session):
 @nox.session(python="pypy")
 def pypy(session):
     session.install(*TEST_DEPENDENCIES)
+    session.install(*ASYNC_DEPENDENCIES)
     session.install(".")
     session.run(
         "pytest",

tests/compute_engine/test__metadata.py

@@ -155,6 +155,49 @@ def test_get_success_text():
     assert result == data
 
 
+def test_get_success_params():
+    data = "foobar"
+    request = make_request(data, headers={"content-type": "text/plain"})
+    params = {"recursive": "true"}
+
+    result = _metadata.get(request, PATH, params=params)
+
+    request.assert_called_once_with(
+        method="GET",
+        url=_metadata._METADATA_ROOT + PATH + "?recursive=true",
+        headers=_metadata._METADATA_HEADERS,
+    )
+    assert result == data
+
+
+def test_get_success_recursive_and_params():
+    data = "foobar"
+    request = make_request(data, headers={"content-type": "text/plain"})
+    params = {"recursive": "false"}
+    result = _metadata.get(request, PATH, recursive=True, params=params)
+
+    request.assert_called_once_with(
+        method="GET",
+        url=_metadata._METADATA_ROOT + PATH + "?recursive=true",
+        headers=_metadata._METADATA_HEADERS,
+    )
+    assert result == data
+
+
+def test_get_success_recursive():
+    data = "foobar"
+    request = make_request(data, headers={"content-type": "text/plain"})
+
+    result = _metadata.get(request, PATH, recursive=True)
+
+    request.assert_called_once_with(
+        method="GET",
+        url=_metadata._METADATA_ROOT + PATH + "?recursive=true",
+        headers=_metadata._METADATA_HEADERS,
+    )
+    assert result == data
+
+
 def test_get_success_custom_root_new_variable():
     request = make_request("{}", headers={"content-type": "application/json"})