Intermediate changes

commit_hash:cc4365f5a0e443b92d87079a9c91e77fea2ddcaf

28 files changed, 1531 insertions, 146 deletions

diff --git a/contrib/python/google-auth/py3/.dist-info/METADATA b/contrib/python/google-auth/py3/.dist-info/METADATA
index 26b8a4974a..261e2a0276 100644
--- a/contrib/python/google-auth/py3/.dist-info/METADATA
+++ b/contrib/python/google-auth/py3/.dist-info/METADATA
@@ -1,6 +1,6 @@
 Metadata-Version: 2.1
 Name: google-auth
-Version: 2.34.0
+Version: 2.35.0
 Summary: Google Authentication Library
 Home-page: https://github.com/googleapis/google-auth-library-python
 Author: Google Cloud Platform
diff --git a/contrib/python/google-auth/py3/google/auth/_default.py b/contrib/python/google-auth/py3/google/auth/_default.py
index 63009dfb86..7bbcf85914 100644
--- a/contrib/python/google-auth/py3/google/auth/_default.py
+++ b/contrib/python/google-auth/py3/google/auth/_default.py
@@ -237,6 +237,7 @@ def _get_gcloud_sdk_credentials(quota_project_id=None):
     credentials, project_id = load_credentials_from_file(
         credentials_filename, quota_project_id=quota_project_id
     )
+    credentials._cred_file_path = credentials_filename
 
     if not project_id:
         project_id = _cloud_sdk.get_project_id()
@@ -270,6 +271,7 @@ def _get_explicit_environ_credentials(quota_project_id=None):
         credentials, project_id = load_credentials_from_file(
             os.environ[environment_vars.CREDENTIALS], quota_project_id=quota_project_id
         )
+        credentials._cred_file_path = f"{explicit_file} file via the GOOGLE_APPLICATION_CREDENTIALS environment variable"
 
         return credentials, project_id
 
diff --git a/contrib/python/google-auth/py3/google/auth/_exponential_backoff.py b/contrib/python/google-auth/py3/google/auth/_exponential_backoff.py
index 04f9f97641..89853448f9 100644
--- a/contrib/python/google-auth/py3/google/auth/_exponential_backoff.py
+++ b/contrib/python/google-auth/py3/google/auth/_exponential_backoff.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import random
 import time
 
@@ -38,9 +39,8 @@ an HTTP request.
 """
 
 
-class ExponentialBackoff:
-    """An exponential backoff iterator. This can be used in a for loop to
-    perform requests with exponential backoff.
+class _BaseExponentialBackoff:
+    """An exponential backoff iterator base class.
 
     Args:
         total_attempts Optional[int]:
@@ -84,9 +84,40 @@ class ExponentialBackoff:
         self._multiplier = multiplier
         self._backoff_count = 0
 
-    def __iter__(self):
+    @property
+    def total_attempts(self):
+        """The total amount of backoff attempts that will be made."""
+        return self._total_attempts
+
+    @property
+    def backoff_count(self):
+        """The current amount of backoff attempts that have been made."""
+        return self._backoff_count
+
+    def _reset(self):
         self._backoff_count = 0
         self._current_wait_in_seconds = self._initial_wait_seconds
+
+    def _calculate_jitter(self):
+        jitter_variance = self._current_wait_in_seconds * self._randomization_factor
+        jitter = random.uniform(
+            self._current_wait_in_seconds - jitter_variance,
+            self._current_wait_in_seconds + jitter_variance,
+        )
+
+        return jitter
+
+
+class ExponentialBackoff(_BaseExponentialBackoff):
+    """An exponential backoff iterator. This can be used in a for loop to
+    perform requests with exponential backoff.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(ExponentialBackoff, self).__init__(*args, **kwargs)
+
+    def __iter__(self):
+        self._reset()
         return self
 
     def __next__(self):
@@ -97,23 +128,37 @@ class ExponentialBackoff:
         if self._backoff_count <= 1:
             return self._backoff_count
 
-        jitter_variance = self._current_wait_in_seconds * self._randomization_factor
-        jitter = random.uniform(
-            self._current_wait_in_seconds - jitter_variance,
-            self._current_wait_in_seconds + jitter_variance,
-        )
+        jitter = self._calculate_jitter()
 
         time.sleep(jitter)
 
         self._current_wait_in_seconds *= self._multiplier
         return self._backoff_count
 
-    @property
-    def total_attempts(self):
-        """The total amount of backoff attempts that will be made."""
-        return self._total_attempts
 
-    @property
-    def backoff_count(self):
-        """The current amount of backoff attempts that have been made."""
+class AsyncExponentialBackoff(_BaseExponentialBackoff):
+    """An async exponential backoff iterator. This can be used in a for loop to
+    perform async requests with exponential backoff.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(AsyncExponentialBackoff, self).__init__(*args, **kwargs)
+
+    def __aiter__(self):
+        self._reset()
+        return self
+
+    async def __anext__(self):
+        if self._backoff_count >= self._total_attempts:
+            raise StopAsyncIteration
+        self._backoff_count += 1
+
+        if self._backoff_count <= 1:
+            return self._backoff_count
+
+        jitter = self._calculate_jitter()
+
+        await asyncio.sleep(jitter)
+
+        self._current_wait_in_seconds *= self._multiplier
         return self._backoff_count
diff --git a/contrib/python/google-auth/py3/google/auth/aio/transport/__init__.py b/contrib/python/google-auth/py3/google/auth/aio/transport/__init__.py
new file mode 100644
index 0000000000..166a3be509
--- /dev/null
+++ b/contrib/python/google-auth/py3/google/auth/aio/transport/__init__.py
@@ -0,0 +1,144 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Transport - Asynchronous HTTP client library support.
+
+:mod:`google.auth.aio` is designed to work with various asynchronous client libraries such
+as aiohttp. In order to work across these libraries with different
+interfaces some abstraction is needed.
+
+This module provides two interfaces that are implemented by transport adapters
+to support HTTP libraries. :class:`Request` defines the interface expected by
+:mod:`google.auth` to make asynchronous requests. :class:`Response` defines the interface
+for the return value of :class:`Request`.
+"""
+
+import abc
+from typing import AsyncGenerator, Mapping, Optional
+
+import google.auth.transport
+
+
+_DEFAULT_TIMEOUT_SECONDS = 180
+
+DEFAULT_RETRYABLE_STATUS_CODES = google.auth.transport.DEFAULT_RETRYABLE_STATUS_CODES
+"""Sequence[int]:  HTTP status codes indicating a request can be retried.
+"""
+
+
+DEFAULT_MAX_RETRY_ATTEMPTS = 3
+"""int: How many times to retry a request."""
+
+
+class Response(metaclass=abc.ABCMeta):
+    """Asynchronous HTTP Response Interface."""
+
+    @property
+    @abc.abstractmethod
+    def status_code(self) -> int:
+        """
+        The HTTP response status code.
+
+        Returns:
+            int: The HTTP response status code.
+
+        """
+        raise NotImplementedError("status_code must be implemented.")
+
+    @property
+    @abc.abstractmethod
+    def headers(self) -> Mapping[str, str]:
+        """The HTTP response headers.
+
+        Returns:
+            Mapping[str, str]: The HTTP response headers.
+        """
+        raise NotImplementedError("headers must be implemented.")
+
+    @abc.abstractmethod
+    async def content(self, chunk_size: int) -> AsyncGenerator[bytes, None]:
+        """The raw response content.
+
+        Args:
+            chunk_size (int): The size of each chunk.
+
+        Yields:
+            AsyncGenerator[bytes, None]: An asynchronous generator yielding
+            response chunks as bytes.
+        """
+        raise NotImplementedError("content must be implemented.")
+
+    @abc.abstractmethod
+    async def read(self) -> bytes:
+        """Read the entire response content as bytes.
+
+        Returns:
+            bytes: The entire response content.
+        """
+        raise NotImplementedError("read must be implemented.")
+
+    @abc.abstractmethod
+    async def close(self):
+        """Close the response after it is fully consumed to resource."""
+        raise NotImplementedError("close must be implemented.")
+
+
+class Request(metaclass=abc.ABCMeta):
+    """Interface for a callable that makes HTTP requests.
+
+    Specific transport implementations should provide an implementation of
+    this that adapts their specific request / response API.
+
+    .. automethod:: __call__
+    """
+
+    @abc.abstractmethod
+    async def __call__(
+        self,
+        url: str,
+        method: str,
+        body: Optional[bytes],
+        headers: Optional[Mapping[str, str]],
+        timeout: float,
+        **kwargs
+    ) -> Response:
+        """Make an HTTP request.
+
+        Args:
+            url (str): The URI to be requested.
+            method (str): The HTTP method to use for the request. Defaults
+                to 'GET'.
+            body (Optional[bytes]): The payload / body in HTTP request.
+            headers (Mapping[str, str]): Request headers.
+            timeout (float): The number of seconds to wait for a
+                response from the server. If not specified or if None, the
+                transport-specific default timeout will be used.
+            kwargs: Additional arguments passed on to the transport's
+                request method.
+
+        Returns:
+            google.auth.aio.transport.Response: The HTTP response.
+
+        Raises:
+            google.auth.exceptions.TransportError: If any exception occurred.
+        """
+        # pylint: disable=redundant-returns-doc, missing-raises-doc
+        # (pylint doesn't play well with abstract docstrings.)
+        raise NotImplementedError("__call__ must be implemented.")
+
+    async def close(self) -> None:
+        """
+        Close the underlying session.
+        """
+        raise NotImplementedError("close must be implemented.")
diff --git a/contrib/python/google-auth/py3/google/auth/aio/transport/aiohttp.py b/contrib/python/google-auth/py3/google/auth/aio/transport/aiohttp.py
new file mode 100644
index 0000000000..074d1491c7
--- /dev/null
+++ b/contrib/python/google-auth/py3/google/auth/aio/transport/aiohttp.py
@@ -0,0 +1,184 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Transport adapter for Asynchronous HTTP Requests based on aiohttp.
+"""
+
+import asyncio
+from typing import AsyncGenerator, Mapping, Optional
+
+try:
+    import aiohttp  # type: ignore
+except ImportError as caught_exc:  # pragma: NO COVER
+    raise ImportError(
+        "The aiohttp library is not installed from please install the aiohttp package to use the aiohttp transport."
+    ) from caught_exc
+
+from google.auth import _helpers
+from google.auth import exceptions
+from google.auth.aio import transport
+
+
+class Response(transport.Response):
+    """
+    Represents an HTTP response and its data. It is returned by ``google.auth.aio.transport.sessions.AsyncAuthorizedSession``.
+
+    Args:
+        response (aiohttp.ClientResponse): An instance of aiohttp.ClientResponse.
+
+    Attributes:
+        status_code (int): The HTTP status code of the response.
+        headers (Mapping[str, str]): The HTTP headers of the response.
+    """
+
+    def __init__(self, response: aiohttp.ClientResponse):
+        self._response = response
+
+    @property
+    @_helpers.copy_docstring(transport.Response)
+    def status_code(self) -> int:
+        return self._response.status
+
+    @property
+    @_helpers.copy_docstring(transport.Response)
+    def headers(self) -> Mapping[str, str]:
+        return {key: value for key, value in self._response.headers.items()}
+
+    @_helpers.copy_docstring(transport.Response)
+    async def content(self, chunk_size: int = 1024) -> AsyncGenerator[bytes, None]:
+        try:
+            async for chunk in self._response.content.iter_chunked(
+                chunk_size
+            ):  # pragma: no branch
+                yield chunk
+        except aiohttp.ClientPayloadError as exc:
+            raise exceptions.ResponseError(
+                "Failed to read from the payload stream."
+            ) from exc
+
+    @_helpers.copy_docstring(transport.Response)
+    async def read(self) -> bytes:
+        try:
+            return await self._response.read()
+        except aiohttp.ClientResponseError as exc:
+            raise exceptions.ResponseError("Failed to read the response body.") from exc
+
+    @_helpers.copy_docstring(transport.Response)
+    async def close(self):
+        self._response.close()
+
+
+class Request(transport.Request):
+    """Asynchronous Requests request adapter.
+
+    This class is used internally for making requests using aiohttp
+    in a consistent way. If you use :class:`google.auth.aio.transport.sessions.AsyncAuthorizedSession`
+    you do not need to construct or use this class directly.
+
+    This class can be useful if you want to configure a Request callable
+    with a custom ``aiohttp.ClientSession`` in :class:`AuthorizedSession` or if
+    you want to manually refresh a :class:`~google.auth.aio.credentials.Credentials` instance::
+
+        import aiohttp
+        import google.auth.aio.transport.aiohttp
+
+        # Default example:
+        request = google.auth.aio.transport.aiohttp.Request()
+        await credentials.refresh(request)
+
+        # Custom aiohttp Session Example:
+        session = session=aiohttp.ClientSession(auto_decompress=False)
+        request = google.auth.aio.transport.aiohttp.Request(session=session)
+        auth_sesion = google.auth.aio.transport.sessions.AsyncAuthorizedSession(auth_request=request)
+
+    Args:
+        session (aiohttp.ClientSession): An instance :class:`aiohttp.ClientSession` used
+            to make HTTP requests. If not specified, a session will be created.
+
+    .. automethod:: __call__
+    """
+
+    def __init__(self, session: aiohttp.ClientSession = None):
+        self._session = session
+        self._closed = False
+
+    async def __call__(
+        self,
+        url: str,
+        method: str = "GET",
+        body: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        """
+        Make an HTTP request using aiohttp.
+
+        Args:
+            url (str): The URL to be requested.
+            method (Optional[str]):
+                The HTTP method to use for the request. Defaults to 'GET'.
+            body (Optional[bytes]):
+                The payload or body in HTTP request.
+            headers (Optional[Mapping[str, str]]):
+                Request headers.
+            timeout (float): The number of seconds to wait for a
+                response from the server. If not specified or if None, the
+                requests default timeout will be used.
+            kwargs: Additional arguments passed through to the underlying
+                aiohttp :meth:`aiohttp.Session.request` method.
+
+        Returns:
+            google.auth.aio.transport.Response: The HTTP response.
+
+        Raises:
+            - google.auth.exceptions.TransportError: If the request fails or if the session is closed.
+            - google.auth.exceptions.TimeoutError: If the request times out.
+        """
+
+        try:
+            if self._closed:
+                raise exceptions.TransportError("session is closed.")
+
+            if not self._session:
+                self._session = aiohttp.ClientSession()
+
+            client_timeout = aiohttp.ClientTimeout(total=timeout)
+            response = await self._session.request(
+                method,
+                url,
+                data=body,
+                headers=headers,
+                timeout=client_timeout,
+                **kwargs,
+            )
+            return Response(response)
+
+        except aiohttp.ClientError as caught_exc:
+            client_exc = exceptions.TransportError(f"Failed to send request to {url}.")
+            raise client_exc from caught_exc
+
+        except asyncio.TimeoutError as caught_exc:
+            timeout_exc = exceptions.TimeoutError(
+                f"Request timed out after {timeout} seconds."
+            )
+            raise timeout_exc from caught_exc
+
+    async def close(self) -> None:
+        """
+        Close the underlying aiohttp session to release the acquired resources.
+        """
+        if not self._closed and self._session:
+            await self._session.close()
+        self._closed = True
diff --git a/contrib/python/google-auth/py3/google/auth/aio/transport/sessions.py b/contrib/python/google-auth/py3/google/auth/aio/transport/sessions.py
new file mode 100644
index 0000000000..fea7cbbb2c
--- /dev/null
+++ b/contrib/python/google-auth/py3/google/auth/aio/transport/sessions.py
@@ -0,0 +1,268 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+from contextlib import asynccontextmanager
+import functools
+import time
+from typing import Mapping, Optional
+
+from google.auth import _exponential_backoff, exceptions
+from google.auth.aio import transport
+from google.auth.aio.credentials import Credentials
+from google.auth.exceptions import TimeoutError
+
+try:
+    from google.auth.aio.transport.aiohttp import Request as AiohttpRequest
+
+    AIOHTTP_INSTALLED = True
+except ImportError:  # pragma: NO COVER
+    AIOHTTP_INSTALLED = False
+
+
+@asynccontextmanager
+async def timeout_guard(timeout):
+    """
+    timeout_guard is an asynchronous context manager to apply a timeout to an asynchronous block of code.
+
+    Args:
+        timeout (float): The time in seconds before the context manager times out.
+
+    Raises:
+        google.auth.exceptions.TimeoutError: If the code within the context exceeds the provided timeout.
+
+    Usage:
+        async with timeout_guard(10) as with_timeout:
+            await with_timeout(async_function())
+    """
+    start = time.monotonic()
+    total_timeout = timeout
+
+    def _remaining_time():
+        elapsed = time.monotonic() - start
+        remaining = total_timeout - elapsed
+        if remaining <= 0:
+            raise TimeoutError(
+                f"Context manager exceeded the configured timeout of {total_timeout}s."
+            )
+        return remaining
+
+    async def with_timeout(coro):
+        try:
+            remaining = _remaining_time()
+            response = await asyncio.wait_for(coro, remaining)
+            return response
+        except (asyncio.TimeoutError, TimeoutError) as e:
+            raise TimeoutError(
+                f"The operation {coro} exceeded the configured timeout of {total_timeout}s."
+            ) from e
+
+    try:
+        yield with_timeout
+
+    finally:
+        _remaining_time()
+
+
+class AsyncAuthorizedSession:
+    """This is an asynchronous implementation of :class:`google.auth.requests.AuthorizedSession` class.
+    We utilize an instance of a class that implements :class:`google.auth.aio.transport.Request` configured
+    by the caller or otherwise default to `google.auth.aio.transport.aiohttp.Request` if the external aiohttp
+    package is installed.
+
+    A Requests Session class with credentials.
+
+    This class is used to perform asynchronous requests to API endpoints that require
+    authorization::
+
+        import aiohttp
+        from google.auth.aio.transport import sessions
+
+        async with sessions.AsyncAuthorizedSession(credentials) as authed_session:
+            response = await authed_session.request(
+                'GET', 'https://www.googleapis.com/storage/v1/b')
+
+    The underlying :meth:`request` implementation handles adding the
+    credentials' headers to the request and refreshing credentials as needed.
+
+    Args:
+        credentials (google.auth.aio.credentials.Credentials):
+            The credentials to add to the request.
+        auth_request (Optional[google.auth.aio.transport.Request]):
+            An instance of a class that implements
+            :class:`~google.auth.aio.transport.Request` used to make requests
+            and refresh credentials. If not passed,
+            an instance of :class:`~google.auth.aio.transport.aiohttp.Request`
+            is created.
+
+    Raises:
+        - google.auth.exceptions.TransportError: If `auth_request` is `None`
+            and the external package `aiohttp` is not installed.
+        - google.auth.exceptions.InvalidType: If the provided credentials are
+            not of type `google.auth.aio.credentials.Credentials`.
+    """
+
+    def __init__(
+        self, credentials: Credentials, auth_request: Optional[transport.Request] = None
+    ):
+        if not isinstance(credentials, Credentials):
+            raise exceptions.InvalidType(
+                f"The configured credentials of type {type(credentials)} are invalid and must be of type `google.auth.aio.credentials.Credentials`"
+            )
+        self._credentials = credentials
+        _auth_request = auth_request
+        if not _auth_request and AIOHTTP_INSTALLED:
+            _auth_request = AiohttpRequest()
+        if _auth_request is None:
+            raise exceptions.TransportError(
+                "`auth_request` must either be configured or the external package `aiohttp` must be installed to use the default value."
+            )
+        self._auth_request = _auth_request
+
+    async def request(
+        self,
+        method: str,
+        url: str,
+        data: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        max_allowed_time: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        """
+        Args:
+                method (str): The http method used to make the request.
+                url (str): The URI to be requested.
+                data (Optional[bytes]): The payload or body in HTTP request.
+                headers (Optional[Mapping[str, str]]): Request headers.
+                timeout (float):
+                The amount of time in seconds to wait for the server response
+                with each individual request.
+                max_allowed_time (float):
+                If the method runs longer than this, a ``Timeout`` exception is
+                automatically raised. Unlike the ``timeout`` parameter, this
+                value applies to the total method execution time, even if
+                multiple requests are made under the hood.
+
+                Mind that it is not guaranteed that the timeout error is raised
+                at ``max_allowed_time``. It might take longer, for example, if
+                an underlying request takes a lot of time, but the request
+                itself does not timeout, e.g. if a large file is being
+                transmitted. The timout error will be raised after such
+                request completes.
+
+        Returns:
+                google.auth.aio.transport.Response: The HTTP response.
+
+        Raises:
+                google.auth.exceptions.TimeoutError: If the method does not complete within
+                the configured `max_allowed_time` or the request exceeds the configured
+                `timeout`.
+        """
+
+        retries = _exponential_backoff.AsyncExponentialBackoff(
+            total_attempts=transport.DEFAULT_MAX_RETRY_ATTEMPTS
+        )
+        async with timeout_guard(max_allowed_time) as with_timeout:
+            await with_timeout(
+                # Note: before_request will attempt to refresh credentials if expired.
+                self._credentials.before_request(
+                    self._auth_request, method, url, headers
+                )
+            )
+            # Workaround issue in python 3.9 related to code coverage by adding `# pragma: no branch`
+            # See https://github.com/googleapis/gapic-generator-python/pull/1174#issuecomment-1025132372
+            async for _ in retries:  # pragma: no branch
+                response = await with_timeout(
+                    self._auth_request(url, method, data, headers, timeout, **kwargs)
+                )
+                if response.status_code not in transport.DEFAULT_RETRYABLE_STATUS_CODES:
+                    break
+        return response
+
+    @functools.wraps(request)
+    async def get(
+        self,
+        url: str,
+        data: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        max_allowed_time: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        return await self.request(
+            "GET", url, data, headers, max_allowed_time, timeout, **kwargs
+        )
+
+    @functools.wraps(request)
+    async def post(
+        self,
+        url: str,
+        data: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        max_allowed_time: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        return await self.request(
+            "POST", url, data, headers, max_allowed_time, timeout, **kwargs
+        )
+
+    @functools.wraps(request)
+    async def put(
+        self,
+        url: str,
+        data: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        max_allowed_time: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        return await self.request(
+            "PUT", url, data, headers, max_allowed_time, timeout, **kwargs
+        )
+
+    @functools.wraps(request)
+    async def patch(
+        self,
+        url: str,
+        data: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        max_allowed_time: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        return await self.request(
+            "PATCH", url, data, headers, max_allowed_time, timeout, **kwargs
+        )
+
+    @functools.wraps(request)
+    async def delete(
+        self,
+        url: str,
+        data: Optional[bytes] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        max_allowed_time: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        timeout: float = transport._DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> transport.Response:
+        return await self.request(
+            "DELETE", url, data, headers, max_allowed_time, timeout, **kwargs
+        )
+
+    async def close(self) -> None:
+        """
+        Close the underlying auth request session.
+        """
+        await self._auth_request.close()
diff --git a/contrib/python/google-auth/py3/google/auth/compute_engine/credentials.py b/contrib/python/google-auth/py3/google/auth/compute_engine/credentials.py
index 008b991bb9..f0126c0a80 100644
--- a/contrib/python/google-auth/py3/google/auth/compute_engine/credentials.py
+++ b/contrib/python/google-auth/py3/google/auth/compute_engine/credentials.py
@@ -157,6 +157,14 @@ class Credentials(
         self._universe_domain_cached = True
         return self._universe_domain
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        return {
+            "credential_source": "metadata server",
+            "credential_type": "VM credentials",
+            "principal": self.service_account_email,
+        }
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
         creds = self.__class__(
diff --git a/contrib/python/google-auth/py3/google/auth/credentials.py b/contrib/python/google-auth/py3/google/auth/credentials.py
index e31930311b..2c67e04432 100644
--- a/contrib/python/google-auth/py3/google/auth/credentials.py
+++ b/contrib/python/google-auth/py3/google/auth/credentials.py
@@ -128,6 +128,17 @@ class Credentials(_BaseCredentials):
         """The universe domain value."""
         return self._universe_domain
 
+    def get_cred_info(self):
+        """The credential information JSON.
+
+        The credential information will be added to auth related error messages
+        by client library.
+
+        Returns:
+            Mapping[str, str]: The credential information JSON.
+        """
+        return None
+
     @abc.abstractmethod
     def refresh(self, request):
         """Refreshes the access token.
diff --git a/contrib/python/google-auth/py3/google/auth/exceptions.py b/contrib/python/google-auth/py3/google/auth/exceptions.py
index fcbe61b746..feb9f7411e 100644
--- a/contrib/python/google-auth/py3/google/auth/exceptions.py
+++ b/contrib/python/google-auth/py3/google/auth/exceptions.py
@@ -98,3 +98,11 @@ class InvalidType(DefaultCredentialsError, TypeError):
 
 class OSError(DefaultCredentialsError, EnvironmentError):
     """Used to wrap EnvironmentError(OSError after python3.3)."""
+
+
+class TimeoutError(GoogleAuthError):
+    """Used to indicate a timeout error occurred during an HTTP request."""
+
+
+class ResponseError(GoogleAuthError):
+    """Used to indicate an error occurred when reading an HTTP response."""
diff --git a/contrib/python/google-auth/py3/google/auth/external_account.py b/contrib/python/google-auth/py3/google/auth/external_account.py
index df0511f255..161e6c50ce 100644
--- a/contrib/python/google-auth/py3/google/auth/external_account.py
+++ b/contrib/python/google-auth/py3/google/auth/external_account.py
@@ -186,6 +186,7 @@ class Credentials(
         self._supplier_context = SupplierContext(
             self._subject_token_type, self._audience
         )
+        self._cred_file_path = None
 
         if not self.is_workforce_pool and self._workforce_pool_user_project:
             # Workload identity pools do not support workforce pool user projects.
@@ -321,11 +322,24 @@ class Credentials(
 
         return self._token_info_url
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            cred_info_json = {
+                "credential_source": self._cred_file_path,
+                "credential_type": "external account credentials",
+            }
+            if self.service_account_email:
+                cred_info_json["principal"] = self.service_account_email
+            return cred_info_json
+        return None
+
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes, default_scopes=None):
         kwargs = self._constructor_args()
         kwargs.update(scopes=scopes, default_scopes=default_scopes)
         scoped = self.__class__(**kwargs)
+        scoped._cred_file_path = self._cred_file_path
         scoped._metrics_options = self._metrics_options
         return scoped
 
@@ -442,30 +456,31 @@ class Credentials(
 
             self.expiry = now + lifetime
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
-        # Return copy of instance with the provided quota project ID.
+    def _make_copy(self):
         kwargs = self._constructor_args()
-        kwargs.update(quota_project_id=quota_project_id)
         new_cred = self.__class__(**kwargs)
+        new_cred._cred_file_path = self._cred_file_path
         new_cred._metrics_options = self._metrics_options
         return new_cred
 
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        # Return copy of instance with the provided quota project ID.
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
+
     @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
     def with_token_uri(self, token_uri):
-        kwargs = self._constructor_args()
-        kwargs.update(token_url=token_uri)
-        new_cred = self.__class__(**kwargs)
-        new_cred._metrics_options = self._metrics_options
-        return new_cred
+        cred = self._make_copy()
+        cred._token_url = token_uri
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-        kwargs = self._constructor_args()
-        kwargs.update(universe_domain=universe_domain)
-        new_cred = self.__class__(**kwargs)
-        new_cred._metrics_options = self._metrics_options
-        return new_cred
+        cred = self._make_copy()
+        cred._universe_domain = universe_domain
+        return cred
 
     def _should_initialize_impersonated_credentials(self):
         return (
diff --git a/contrib/python/google-auth/py3/google/auth/external_account_authorized_user.py b/contrib/python/google-auth/py3/google/auth/external_account_authorized_user.py
index f73387172c..4d0c3c6806 100644
--- a/contrib/python/google-auth/py3/google/auth/external_account_authorized_user.py
+++ b/contrib/python/google-auth/py3/google/auth/external_account_authorized_user.py
@@ -120,6 +120,7 @@ class Credentials(
         self._quota_project_id = quota_project_id
         self._scopes = scopes
         self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
+        self._cred_file_path = None
 
         if not self.valid and not self.can_refresh:
             raise exceptions.InvalidOperation(
@@ -290,23 +291,38 @@ class Credentials(
     def _make_sts_request(self, request):
         return self._sts_client.refresh_token(request, self._refresh_token)
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            return {
+                "credential_source": self._cred_file_path,
+                "credential_type": "external account authorized user credentials",
+            }
+        return None
+
+    def _make_copy(self):
+        kwargs = self.constructor_args()
+        cred = self.__class__(**kwargs)
+        cred._cred_file_path = self._cred_file_path
+        return cred
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
-        kwargs = self.constructor_args()
-        kwargs.update(quota_project_id=quota_project_id)
-        return self.__class__(**kwargs)
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
     def with_token_uri(self, token_uri):
-        kwargs = self.constructor_args()
-        kwargs.update(token_url=token_uri)
-        return self.__class__(**kwargs)
+        cred = self._make_copy()
+        cred._token_url = token_uri
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-        kwargs = self.constructor_args()
-        kwargs.update(universe_domain=universe_domain)
-        return self.__class__(**kwargs)
+        cred = self._make_copy()
+        cred._universe_domain = universe_domain
+        return cred
 
     @classmethod
     def from_info(cls, info, **kwargs):
diff --git a/contrib/python/google-auth/py3/google/auth/impersonated_credentials.py b/contrib/python/google-auth/py3/google/auth/impersonated_credentials.py
index 3c6f8712a9..c42a936433 100644
--- a/contrib/python/google-auth/py3/google/auth/impersonated_credentials.py
+++ b/contrib/python/google-auth/py3/google/auth/impersonated_credentials.py
@@ -226,6 +226,7 @@ class Credentials(
         self.expiry = _helpers.utcnow()
         self._quota_project_id = quota_project_id
         self._iam_endpoint_override = iam_endpoint_override
+        self._cred_file_path = None
 
     def _metric_header_for_usage(self):
         return metrics.CRED_TYPE_SA_IMPERSONATE
@@ -316,29 +317,40 @@ class Credentials(
     def requires_scopes(self):
         return not self._target_scopes
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
-        return self.__class__(
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            return {
+                "credential_source": self._cred_file_path,
+                "credential_type": "impersonated credentials",
+                "principal": self._target_principal,
+            }
+        return None
+
+    def _make_copy(self):
+        cred = self.__class__(
             self._source_credentials,
             target_principal=self._target_principal,
             target_scopes=self._target_scopes,
             delegates=self._delegates,
             lifetime=self._lifetime,
-            quota_project_id=quota_project_id,
+            quota_project_id=self._quota_project_id,
             iam_endpoint_override=self._iam_endpoint_override,
         )
+        cred._cred_file_path = self._cred_file_path
+        return cred
+
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
 
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes, default_scopes=None):
-        return self.__class__(
-            self._source_credentials,
-            target_principal=self._target_principal,
-            target_scopes=scopes or default_scopes,
-            delegates=self._delegates,
-            lifetime=self._lifetime,
-            quota_project_id=self._quota_project_id,
-            iam_endpoint_override=self._iam_endpoint_override,
-        )
+        cred = self._make_copy()
+        cred._target_scopes = scopes or default_scopes
+        return cred
 
 
 class IDTokenCredentials(credentials.CredentialsWithQuotaProject):
diff --git a/contrib/python/google-auth/py3/google/auth/version.py b/contrib/python/google-auth/py3/google/auth/version.py
index 297e18a45f..6610120c69 100644
--- a/contrib/python/google-auth/py3/google/auth/version.py
+++ b/contrib/python/google-auth/py3/google/auth/version.py
@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "2.34.0"
+__version__ = "2.35.0"
diff --git a/contrib/python/google-auth/py3/google/oauth2/credentials.py b/contrib/python/google-auth/py3/google/oauth2/credentials.py
index 5ca00d4c5a..6e158089f3 100644
--- a/contrib/python/google-auth/py3/google/oauth2/credentials.py
+++ b/contrib/python/google-auth/py3/google/oauth2/credentials.py
@@ -50,6 +50,9 @@ _LOGGER = logging.getLogger(__name__)
 # The Google OAuth 2.0 token endpoint. Used for authorized user credentials.
 _GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
 
+# The Google OAuth 2.0 token info endpoint. Used for getting token info JSON from access tokens.
+_GOOGLE_OAUTH2_TOKEN_INFO_ENDPOINT = "https://oauth2.googleapis.com/tokeninfo"
+
 
 class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaProject):
     """Credentials using OAuth 2.0 access and refresh tokens.
@@ -151,6 +154,7 @@ class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaPr
         self._trust_boundary = trust_boundary
         self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
         self._account = account or ""
+        self._cred_file_path = None
 
     def __getstate__(self):
         """A __getstate__ method must exist for the __setstate__ to be called
@@ -189,6 +193,7 @@ class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaPr
         self._universe_domain = (
             d.get("_universe_domain") or credentials.DEFAULT_UNIVERSE_DOMAIN
         )
+        self._cred_file_path = d.get("_cred_file_path")
         # The refresh_handler setter should be used to repopulate this.
         self._refresh_handler = None
         self._refresh_worker = None
@@ -278,10 +283,8 @@ class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaPr
         """str: The user account associated with the credential. If the account is unknown an empty string is returned."""
         return self._account
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
-
-        return self.__class__(
+    def _make_copy(self):
+        cred = self.__class__(
             self.token,
             refresh_token=self.refresh_token,
             id_token=self.id_token,
@@ -291,34 +294,39 @@ class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaPr
             scopes=self.scopes,
             default_scopes=self.default_scopes,
             granted_scopes=self.granted_scopes,
-            quota_project_id=quota_project_id,
+            quota_project_id=self.quota_project_id,
             rapt_token=self.rapt_token,
             enable_reauth_refresh=self._enable_reauth_refresh,
             trust_boundary=self._trust_boundary,
             universe_domain=self._universe_domain,
             account=self._account,
         )
+        cred._cred_file_path = self._cred_file_path
+        return cred
+
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            cred_info = {
+                "credential_source": self._cred_file_path,
+                "credential_type": "user credentials",
+            }
+            if self.account:
+                cred_info["principal"] = self.account
+            return cred_info
+        return None
+
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
     def with_token_uri(self, token_uri):
-
-        return self.__class__(
-            self.token,
-            refresh_token=self.refresh_token,
-            id_token=self.id_token,
-            token_uri=token_uri,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            scopes=self.scopes,
-            default_scopes=self.default_scopes,
-            granted_scopes=self.granted_scopes,
-            quota_project_id=self.quota_project_id,
-            rapt_token=self.rapt_token,
-            enable_reauth_refresh=self._enable_reauth_refresh,
-            trust_boundary=self._trust_boundary,
-            universe_domain=self._universe_domain,
-            account=self._account,
-        )
+        cred = self._make_copy()
+        cred._token_uri = token_uri
+        return cred
 
     def with_account(self, account):
         """Returns a copy of these credentials with a modified account.
@@ -329,45 +337,15 @@ class Credentials(credentials.ReadOnlyScoped, credentials.CredentialsWithQuotaPr
         Returns:
             google.oauth2.credentials.Credentials: A new credentials instance.
         """
-
-        return self.__class__(
-            self.token,
-            refresh_token=self.refresh_token,
-            id_token=self.id_token,
-            token_uri=self._token_uri,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            scopes=self.scopes,
-            default_scopes=self.default_scopes,
-            granted_scopes=self.granted_scopes,
-            quota_project_id=self.quota_project_id,
-            rapt_token=self.rapt_token,
-            enable_reauth_refresh=self._enable_reauth_refresh,
-            trust_boundary=self._trust_boundary,
-            universe_domain=self._universe_domain,
-            account=account,
-        )
+        cred = self._make_copy()
+        cred._account = account
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-
-        return self.__class__(
-            self.token,
-            refresh_token=self.refresh_token,
-            id_token=self.id_token,
-            token_uri=self._token_uri,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            scopes=self.scopes,
-            default_scopes=self.default_scopes,
-            granted_scopes=self.granted_scopes,
-            quota_project_id=self.quota_project_id,
-            rapt_token=self.rapt_token,
-            enable_reauth_refresh=self._enable_reauth_refresh,
-            trust_boundary=self._trust_boundary,
-            universe_domain=universe_domain,
-            account=self._account,
-        )
+        cred = self._make_copy()
+        cred._universe_domain = universe_domain
+        return cred
 
     def _metric_header_for_usage(self):
         return metrics.CRED_TYPE_USER
diff --git a/contrib/python/google-auth/py3/google/oauth2/service_account.py b/contrib/python/google-auth/py3/google/oauth2/service_account.py
index 0e12868f14..98dafa3e38 100644
--- a/contrib/python/google-auth/py3/google/oauth2/service_account.py
+++ b/contrib/python/google-auth/py3/google/oauth2/service_account.py
@@ -173,6 +173,7 @@ class Credentials(
         """
         super(Credentials, self).__init__()
 
+        self._cred_file_path = None
         self._scopes = scopes
         self._default_scopes = default_scopes
         self._signer = signer
@@ -220,7 +221,7 @@ class Credentials(
                 "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
             ),
             trust_boundary=info.get("trust_boundary"),
-            **kwargs
+            **kwargs,
         )
 
     @classmethod
@@ -294,6 +295,7 @@ class Credentials(
             always_use_jwt_access=self._always_use_jwt_access,
             universe_domain=self._universe_domain,
         )
+        cred._cred_file_path = self._cred_file_path
         return cred
 
     @_helpers.copy_docstring(credentials.Scoped)
@@ -503,6 +505,16 @@ class Credentials(
     def signer_email(self):
         return self._service_account_email
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            return {
+                "credential_source": self._cred_file_path,
+                "credential_type": "service account credentials",
+                "principal": self.service_account_email,
+            }
+        return None
+
 
 class IDTokenCredentials(
     credentials.Signing,
diff --git a/contrib/python/google-auth/py3/tests/compute_engine/test_credentials.py b/contrib/python/google-auth/py3/tests/compute_engine/test_credentials.py
index bb29f8c6e2..662210fa41 100644
--- a/contrib/python/google-auth/py3/tests/compute_engine/test_credentials.py
+++ b/contrib/python/google-auth/py3/tests/compute_engine/test_credentials.py
@@ -72,6 +72,13 @@ class TestCredentials(object):
             universe_domain=FAKE_UNIVERSE_DOMAIN,
         )
 
+    def test_get_cred_info(self):
+        assert self.credentials.get_cred_info() == {
+            "credential_source": "metadata server",
+            "credential_type": "VM credentials",
+            "principal": "default",
+        }
+
     def test_default_state(self):
         assert not self.credentials.valid
         # Expiration hasn't been set yet
diff --git a/contrib/python/google-auth/py3/tests/oauth2/test_credentials.py b/contrib/python/google-auth/py3/tests/oauth2/test_credentials.py
index 67b6b9c1ad..a4cac7a463 100644
--- a/contrib/python/google-auth/py3/tests/oauth2/test_credentials.py
+++ b/contrib/python/google-auth/py3/tests/oauth2/test_credentials.py
@@ -72,6 +72,34 @@ class TestCredentials(object):
         assert credentials.rapt_token == self.RAPT_TOKEN
         assert credentials.refresh_handler is None
 
+    def test_get_cred_info(self):
+        credentials = self.make_credentials()
+        credentials._account = "fake-account"
+        assert not credentials.get_cred_info()
+
+        credentials._cred_file_path = "/path/to/file"
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "user credentials",
+            "principal": "fake-account",
+        }
+
+    def test_get_cred_info_no_account(self):
+        credentials = self.make_credentials()
+        assert not credentials.get_cred_info()
+
+        credentials._cred_file_path = "/path/to/file"
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "user credentials",
+        }
+
+    def test__make_copy_get_cred_info(self):
+        credentials = self.make_credentials()
+        credentials._cred_file_path = "/path/to/file"
+        cred_copy = credentials._make_copy()
+        assert cred_copy._cred_file_path == "/path/to/file"
+
     def test_token_usage_metrics(self):
         credentials = self.make_credentials()
         credentials.token = "token"
diff --git a/contrib/python/google-auth/py3/tests/oauth2/test_service_account.py b/contrib/python/google-auth/py3/tests/oauth2/test_service_account.py
index 0dbe316a0f..fe02e828e7 100644
--- a/contrib/python/google-auth/py3/tests/oauth2/test_service_account.py
+++ b/contrib/python/google-auth/py3/tests/oauth2/test_service_account.py
@@ -69,6 +69,23 @@ class TestCredentials(object):
             universe_domain=universe_domain,
         )
 
+    def test_get_cred_info(self):
+        credentials = self.make_credentials()
+        assert not credentials.get_cred_info()
+
+        credentials._cred_file_path = "/path/to/file"
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "service account credentials",
+            "principal": "service-account@example.com",
+        }
+
+    def test__make_copy_get_cred_info(self):
+        credentials = self.make_credentials()
+        credentials._cred_file_path = "/path/to/file"
+        cred_copy = credentials._make_copy()
+        assert cred_copy._cred_file_path == "/path/to/file"
+
     def test_constructor_no_universe_domain(self):
         credentials = service_account.Credentials(
             SIGNER, self.SERVICE_ACCOUNT_EMAIL, self.TOKEN_URI, universe_domain=None
diff --git a/contrib/python/google-auth/py3/tests/test__default.py b/contrib/python/google-auth/py3/tests/test__default.py
index aaf892f6d0..3147d505da 100644
--- a/contrib/python/google-auth/py3/tests/test__default.py
+++ b/contrib/python/google-auth/py3/tests/test__default.py
@@ -884,6 +884,38 @@ def test_default_early_out(unused_get):
 
 
 @mock.patch(
+    "google.auth._default.load_credentials_from_file",
+    return_value=(MOCK_CREDENTIALS, mock.sentinel.project_id),
+    autospec=True,
+)
+def test_default_cred_file_path_env_var(unused_load_cred, monkeypatch):
+    monkeypatch.setenv(environment_vars.CREDENTIALS, "/path/to/file")
+    cred, _ = _default.default()
+    assert (
+        cred._cred_file_path
+        == "/path/to/file file via the GOOGLE_APPLICATION_CREDENTIALS environment variable"
+    )
+
+
+@mock.patch("os.path.isfile", return_value=True, autospec=True)
+@mock.patch(
+    "google.auth._cloud_sdk.get_application_default_credentials_path",
+    return_value="/path/to/adc/file",
+    autospec=True,
+)
+@mock.patch(
+    "google.auth._default.load_credentials_from_file",
+    return_value=(MOCK_CREDENTIALS, mock.sentinel.project_id),
+    autospec=True,
+)
+def test_default_cred_file_path_gcloud(
+    unused_load_cred, unused_get_adc_file, unused_isfile
+):
+    cred, _ = _default.default()
+    assert cred._cred_file_path == "/path/to/adc/file"
+
+
+@mock.patch(
     "google.auth._default._get_explicit_environ_credentials",
     return_value=(MOCK_CREDENTIALS, mock.sentinel.project_id),
     autospec=True,
diff --git a/contrib/python/google-auth/py3/tests/test__exponential_backoff.py b/contrib/python/google-auth/py3/tests/test__exponential_backoff.py
index 95422502b0..b7b6877b2c 100644
--- a/contrib/python/google-auth/py3/tests/test__exponential_backoff.py
+++ b/contrib/python/google-auth/py3/tests/test__exponential_backoff.py
@@ -54,3 +54,44 @@ def test_minimum_total_attempts():
     with pytest.raises(exceptions.InvalidValue):
         _exponential_backoff.ExponentialBackoff(total_attempts=-1)
     _exponential_backoff.ExponentialBackoff(total_attempts=1)
+
+
+@pytest.mark.asyncio
+@mock.patch("asyncio.sleep", return_value=None)
+async def test_exponential_backoff_async(mock_time_async):
+    eb = _exponential_backoff.AsyncExponentialBackoff()
+    curr_wait = eb._current_wait_in_seconds
+    iteration_count = 0
+
+    # Workaround issue in python 3.9 related to code coverage by adding `# pragma: no branch`
+    # See https://github.com/googleapis/gapic-generator-python/pull/1174#issuecomment-1025132372
+    async for attempt in eb:  # pragma: no branch
+        if attempt == 1:
+            assert mock_time_async.call_count == 0
+        else:
+            backoff_interval = mock_time_async.call_args[0][0]
+            jitter = curr_wait * eb._randomization_factor
+
+            assert (curr_wait - jitter) <= backoff_interval <= (curr_wait + jitter)
+            assert attempt == iteration_count + 1
+            assert eb.backoff_count == iteration_count + 1
+            assert eb._current_wait_in_seconds == eb._multiplier ** iteration_count
+
+            curr_wait = eb._current_wait_in_seconds
+        iteration_count += 1
+
+    assert eb.total_attempts == _exponential_backoff._DEFAULT_RETRY_TOTAL_ATTEMPTS
+    assert eb.backoff_count == _exponential_backoff._DEFAULT_RETRY_TOTAL_ATTEMPTS
+    assert iteration_count == _exponential_backoff._DEFAULT_RETRY_TOTAL_ATTEMPTS
+    assert (
+        mock_time_async.call_count
+        == _exponential_backoff._DEFAULT_RETRY_TOTAL_ATTEMPTS - 1
+    )
+
+
+def test_minimum_total_attempts_async():
+    with pytest.raises(exceptions.InvalidValue):
+        _exponential_backoff.AsyncExponentialBackoff(total_attempts=0)
+    with pytest.raises(exceptions.InvalidValue):
+        _exponential_backoff.AsyncExponentialBackoff(total_attempts=-1)
+    _exponential_backoff.AsyncExponentialBackoff(total_attempts=1)
diff --git a/contrib/python/google-auth/py3/tests/test_credentials.py b/contrib/python/google-auth/py3/tests/test_credentials.py
index 8e6bbc9633..e11bcb4e55 100644
--- a/contrib/python/google-auth/py3/tests/test_credentials.py
+++ b/contrib/python/google-auth/py3/tests/test_credentials.py
@@ -52,6 +52,11 @@ def test_credentials_constructor():
     assert not credentials._use_non_blocking_refresh
 
 
+def test_credentials_get_cred_info():
+    credentials = CredentialsImpl()
+    assert not credentials.get_cred_info()
+
+
 def test_with_non_blocking_refresh():
     c = CredentialsImpl()
     c.with_non_blocking_refresh()
diff --git a/contrib/python/google-auth/py3/tests/test_external_account.py b/contrib/python/google-auth/py3/tests/test_external_account.py
index 3c372e6291..bddcb4afa1 100644
--- a/contrib/python/google-auth/py3/tests/test_external_account.py
+++ b/contrib/python/google-auth/py3/tests/test_external_account.py
@@ -275,6 +275,31 @@ class TestCredentials(object):
         assert request_kwargs["headers"] == headers
         assert "body" not in request_kwargs
 
+    def test_get_cred_info(self):
+        credentials = self.make_credentials()
+        assert not credentials.get_cred_info()
+
+        credentials._cred_file_path = "/path/to/file"
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "external account credentials",
+        }
+
+        credentials._service_account_impersonation_url = (
+            self.SERVICE_ACCOUNT_IMPERSONATION_URL
+        )
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "external account credentials",
+            "principal": SERVICE_ACCOUNT_EMAIL,
+        }
+
+    def test__make_copy_get_cred_info(self):
+        credentials = self.make_credentials()
+        credentials._cred_file_path = "/path/to/file"
+        cred_copy = credentials._make_copy()
+        assert cred_copy._cred_file_path == "/path/to/file"
+
     def test_default_state(self):
         credentials = self.make_credentials(
             service_account_impersonation_url=self.SERVICE_ACCOUNT_IMPERSONATION_URL
@@ -469,25 +494,29 @@ class TestCredentials(object):
         with mock.patch.object(
             external_account.Credentials, "__init__", return_value=None
         ) as mock_init:
-            credentials.with_quota_project("project-foo")
+            new_cred = credentials.with_quota_project("project-foo")
 
-        # Confirm with_quota_project initialized the credential with the
-        # expected parameters and quota project ID.
-        mock_init.assert_called_once_with(
-            audience=self.AUDIENCE,
-            subject_token_type=self.SUBJECT_TOKEN_TYPE,
-            token_url=self.TOKEN_URL,
-            token_info_url=self.TOKEN_INFO_URL,
-            credential_source=self.CREDENTIAL_SOURCE,
-            service_account_impersonation_url=self.SERVICE_ACCOUNT_IMPERSONATION_URL,
-            service_account_impersonation_options={"token_lifetime_seconds": 2800},
-            client_id=CLIENT_ID,
-            client_secret=CLIENT_SECRET,
-            quota_project_id="project-foo",
-            scopes=self.SCOPES,
-            default_scopes=["default1"],
-            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
-        )
+            # Confirm with_quota_project initialized the credential with the
+            # expected parameters.
+            mock_init.assert_called_once_with(
+                audience=self.AUDIENCE,
+                subject_token_type=self.SUBJECT_TOKEN_TYPE,
+                token_url=self.TOKEN_URL,
+                token_info_url=self.TOKEN_INFO_URL,
+                credential_source=self.CREDENTIAL_SOURCE,
+                service_account_impersonation_url=self.SERVICE_ACCOUNT_IMPERSONATION_URL,
+                service_account_impersonation_options={"token_lifetime_seconds": 2800},
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                quota_project_id=self.QUOTA_PROJECT_ID,
+                scopes=self.SCOPES,
+                default_scopes=["default1"],
+                universe_domain=DEFAULT_UNIVERSE_DOMAIN,
+            )
+
+            # Confirm with_quota_project sets the correct quota project after
+            # initialization.
+            assert new_cred.quota_project_id == "project-foo"
 
     def test_info(self):
         credentials = self.make_credentials(universe_domain="dummy_universe.com")
diff --git a/contrib/python/google-auth/py3/tests/test_external_account_authorized_user.py b/contrib/python/google-auth/py3/tests/test_external_account_authorized_user.py
index 743ee9c848..93926a1314 100644
--- a/contrib/python/google-auth/py3/tests/test_external_account_authorized_user.py
+++ b/contrib/python/google-auth/py3/tests/test_external_account_authorized_user.py
@@ -83,6 +83,22 @@ class TestCredentials(object):
 
         return request
 
+    def test_get_cred_info(self):
+        credentials = self.make_credentials()
+        assert not credentials.get_cred_info()
+
+        credentials._cred_file_path = "/path/to/file"
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "external account authorized user credentials",
+        }
+
+    def test__make_copy_get_cred_info(self):
+        credentials = self.make_credentials()
+        credentials._cred_file_path = "/path/to/file"
+        cred_copy = credentials._make_copy()
+        assert cred_copy._cred_file_path == "/path/to/file"
+
     def test_default_state(self):
         creds = self.make_credentials()
 
diff --git a/contrib/python/google-auth/py3/tests/test_impersonated_credentials.py b/contrib/python/google-auth/py3/tests/test_impersonated_credentials.py
index 7295bba429..4fb68103a8 100644
--- a/contrib/python/google-auth/py3/tests/test_impersonated_credentials.py
+++ b/contrib/python/google-auth/py3/tests/test_impersonated_credentials.py
@@ -136,6 +136,23 @@ class TestImpersonatedCredentials(object):
             iam_endpoint_override=iam_endpoint_override,
         )
 
+    def test_get_cred_info(self):
+        credentials = self.make_credentials()
+        assert not credentials.get_cred_info()
+
+        credentials._cred_file_path = "/path/to/file"
+        assert credentials.get_cred_info() == {
+            "credential_source": "/path/to/file",
+            "credential_type": "impersonated credentials",
+            "principal": "impersonated@project.iam.gserviceaccount.com",
+        }
+
+    def test__make_copy_get_cred_info(self):
+        credentials = self.make_credentials()
+        credentials._cred_file_path = "/path/to/file"
+        cred_copy = credentials._make_copy()
+        assert cred_copy._cred_file_path == "/path/to/file"
+
     def test_make_from_user_credentials(self):
         credentials = self.make_credentials(
             source_credentials=self.USER_SOURCE_CREDENTIALS
diff --git a/contrib/python/google-auth/py3/tests/transport/aio/test_aiohttp.py b/contrib/python/google-auth/py3/tests/transport/aio/test_aiohttp.py
new file mode 100644
index 0000000000..632abff25a
--- /dev/null
+++ b/contrib/python/google-auth/py3/tests/transport/aio/test_aiohttp.py
@@ -0,0 +1,170 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+
+from aioresponses import aioresponses  # type: ignore
+from mock import AsyncMock, Mock, patch
+import pytest  # type: ignore
+import pytest_asyncio  # type: ignore
+
+from google.auth import exceptions
+import google.auth.aio.transport.aiohttp as auth_aiohttp
+
+
+try:
+    import aiohttp  # type: ignore
+except ImportError as caught_exc:  # pragma: NO COVER
+    raise ImportError(
+        "The aiohttp library is not installed from please install the aiohttp package to use the aiohttp transport."
+    ) from caught_exc
+
+
+@pytest.fixture
+def mock_response():
+    response = Mock()
+    response.status = 200
+    response.headers = {"Content-Type": "application/json", "Content-Length": "100"}
+    mock_iterator = AsyncMock()
+    mock_iterator.__aiter__.return_value = iter(
+        [b"Cavefish ", b"have ", b"no ", b"sight."]
+    )
+    response.content.iter_chunked = lambda chunk_size: mock_iterator
+    response.read = AsyncMock(return_value=b"Cavefish have no sight.")
+    response.close = AsyncMock()
+
+    return auth_aiohttp.Response(response)
+
+
+class TestResponse(object):
+    @pytest.mark.asyncio
+    async def test_response_status_code(self, mock_response):
+        assert mock_response.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_response_headers(self, mock_response):
+        assert mock_response.headers["Content-Type"] == "application/json"
+        assert mock_response.headers["Content-Length"] == "100"
+
+    @pytest.mark.asyncio
+    async def test_response_content(self, mock_response):
+        content = b"".join([chunk async for chunk in mock_response.content()])
+        assert content == b"Cavefish have no sight."
+
+    @pytest.mark.asyncio
+    async def test_response_content_raises_error(self, mock_response):
+        with patch.object(
+            mock_response._response.content,
+            "iter_chunked",
+            side_effect=aiohttp.ClientPayloadError,
+        ):
+            with pytest.raises(exceptions.ResponseError) as exc:
+                [chunk async for chunk in mock_response.content()]
+            exc.match("Failed to read from the payload stream")
+
+    @pytest.mark.asyncio
+    async def test_response_read(self, mock_response):
+        content = await mock_response.read()
+        assert content == b"Cavefish have no sight."
+
+    @pytest.mark.asyncio
+    async def test_response_read_raises_error(self, mock_response):
+        with patch.object(
+            mock_response._response,
+            "read",
+            side_effect=aiohttp.ClientResponseError(None, None),
+        ):
+            with pytest.raises(exceptions.ResponseError) as exc:
+                await mock_response.read()
+            exc.match("Failed to read the response body.")
+
+    @pytest.mark.asyncio
+    async def test_response_close(self, mock_response):
+        await mock_response.close()
+        mock_response._response.close.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_response_content_stream(self, mock_response):
+        itr = mock_response.content().__aiter__()
+        content = []
+        try:
+            while True:
+                chunk = await itr.__anext__()
+                content.append(chunk)
+        except StopAsyncIteration:
+            pass
+        assert b"".join(content) == b"Cavefish have no sight."
+
+
+@pytest.mark.asyncio
+class TestRequest:
+    @pytest_asyncio.fixture
+    async def aiohttp_request(self):
+        request = auth_aiohttp.Request()
+        yield request
+        await request.close()
+
+    async def test_request_call_success(self, aiohttp_request):
+        with aioresponses() as m:
+            mocked_chunks = [b"Cavefish ", b"have ", b"no ", b"sight."]
+            mocked_response = b"".join(mocked_chunks)
+            m.get("http://example.com", status=200, body=mocked_response)
+            response = await aiohttp_request("http://example.com")
+            assert response.status_code == 200
+            assert response.headers == {"Content-Type": "application/json"}
+            content = b"".join([chunk async for chunk in response.content()])
+            assert content == b"Cavefish have no sight."
+
+    async def test_request_call_success_with_provided_session(self):
+        mock_session = aiohttp.ClientSession()
+        request = auth_aiohttp.Request(mock_session)
+        with aioresponses() as m:
+            mocked_chunks = [b"Cavefish ", b"have ", b"no ", b"sight."]
+            mocked_response = b"".join(mocked_chunks)
+            m.get("http://example.com", status=200, body=mocked_response)
+            response = await request("http://example.com")
+            assert response.status_code == 200
+            assert response.headers == {"Content-Type": "application/json"}
+            content = b"".join([chunk async for chunk in response.content()])
+            assert content == b"Cavefish have no sight."
+
+    async def test_request_call_raises_client_error(self, aiohttp_request):
+        with aioresponses() as m:
+            m.get("http://example.com", exception=aiohttp.ClientError)
+
+            with pytest.raises(exceptions.TransportError) as exc:
+                await aiohttp_request("http://example.com/api")
+
+            exc.match("Failed to send request to http://example.com/api.")
+
+    async def test_request_call_raises_timeout_error(self, aiohttp_request):
+        with aioresponses() as m:
+            m.get("http://example.com", exception=asyncio.TimeoutError)
+
+            with pytest.raises(exceptions.TimeoutError) as exc:
+                await aiohttp_request("http://example.com")
+
+            exc.match("Request timed out after 180 seconds.")
+
+    async def test_request_call_raises_transport_error_for_closed_session(
+        self, aiohttp_request
+    ):
+        with aioresponses() as m:
+            m.get("http://example.com", exception=asyncio.TimeoutError)
+            aiohttp_request._closed = True
+            with pytest.raises(exceptions.TransportError) as exc:
+                await aiohttp_request("http://example.com")
+
+            exc.match("session is closed.")
+            aiohttp_request._closed = False
diff --git a/contrib/python/google-auth/py3/tests/transport/aio/test_sessions.py b/contrib/python/google-auth/py3/tests/transport/aio/test_sessions.py
new file mode 100644
index 0000000000..c91a7c40ae
--- /dev/null
+++ b/contrib/python/google-auth/py3/tests/transport/aio/test_sessions.py
@@ -0,0 +1,311 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+from typing import AsyncGenerator
+
+from aioresponses import aioresponses  # type: ignore
+from mock import Mock, patch
+import pytest  # type: ignore
+
+from google.auth.aio.credentials import AnonymousCredentials
+from google.auth.aio.transport import (
+    _DEFAULT_TIMEOUT_SECONDS,
+    DEFAULT_MAX_RETRY_ATTEMPTS,
+    DEFAULT_RETRYABLE_STATUS_CODES,
+    Request,
+    Response,
+    sessions,
+)
+from google.auth.exceptions import InvalidType, TimeoutError, TransportError
+
+
+@pytest.fixture
+async def simple_async_task():
+    return True
+
+
+class MockRequest(Request):
+    def __init__(self, response=None, side_effect=None):
+        self._closed = False
+        self._response = response
+        self._side_effect = side_effect
+        self.call_count = 0
+
+    async def __call__(
+        self,
+        url,
+        method="GET",
+        body=None,
+        headers=None,
+        timeout=_DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ):
+        self.call_count += 1
+        if self._side_effect:
+            raise self._side_effect
+        return self._response
+
+    async def close(self):
+        self._closed = True
+        return None
+
+
+class MockResponse(Response):
+    def __init__(self, status_code, headers=None, content=None):
+        self._status_code = status_code
+        self._headers = headers
+        self._content = content
+        self._close = False
+
+    @property
+    def status_code(self):
+        return self._status_code
+
+    @property
+    def headers(self):
+        return self._headers
+
+    async def read(self) -> bytes:
+        content = await self.content(1024)
+        return b"".join([chunk async for chunk in content])
+
+    async def content(self, chunk_size=None) -> AsyncGenerator:
+        return self._content
+
+    async def close(self) -> None:
+        self._close = True
+
+
+class TestTimeoutGuard(object):
+    default_timeout = 1
+
+    def make_timeout_guard(self, timeout):
+        return sessions.timeout_guard(timeout)
+
+    @pytest.mark.asyncio
+    async def test_timeout_with_simple_async_task_within_bounds(
+        self, simple_async_task
+    ):
+        task = False
+        with patch("time.monotonic", side_effect=[0, 0.25, 0.75]):
+            with patch("asyncio.wait_for", lambda coro, _: coro):
+                async with self.make_timeout_guard(
+                    timeout=self.default_timeout
+                ) as with_timeout:
+                    task = await with_timeout(simple_async_task)
+
+        # Task succeeds.
+        assert task is True
+
+    @pytest.mark.asyncio
+    async def test_timeout_with_simple_async_task_out_of_bounds(
+        self, simple_async_task
+    ):
+        task = False
+        with patch("time.monotonic", side_effect=[0, 1, 1]):
+            with pytest.raises(TimeoutError) as exc:
+                async with self.make_timeout_guard(
+                    timeout=self.default_timeout
+                ) as with_timeout:
+                    task = await with_timeout(simple_async_task)
+
+        # Task does not succeed and the context manager times out i.e. no remaining time left.
+        assert task is False
+        assert exc.match(
+            f"Context manager exceeded the configured timeout of {self.default_timeout}s."
+        )
+
+    @pytest.mark.asyncio
+    async def test_timeout_with_async_task_timing_out_before_context(
+        self, simple_async_task
+    ):
+        task = False
+        with pytest.raises(TimeoutError) as exc:
+            async with self.make_timeout_guard(
+                timeout=self.default_timeout
+            ) as with_timeout:
+                with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError):
+                    task = await with_timeout(simple_async_task)
+
+        # Task does not complete i.e. the operation times out.
+        assert task is False
+        assert exc.match(
+            f"The operation {simple_async_task} exceeded the configured timeout of {self.default_timeout}s."
+        )
+
+
+class TestAsyncAuthorizedSession(object):
+    TEST_URL = "http://example.com/"
+    credentials = AnonymousCredentials()
+
+    @pytest.fixture
+    async def mocked_content(self):
+        content = [b"Cavefish ", b"have ", b"no ", b"sight."]
+        for chunk in content:
+            yield chunk
+
+    @pytest.mark.asyncio
+    async def test_constructor_with_default_auth_request(self):
+        with patch("google.auth.aio.transport.sessions.AIOHTTP_INSTALLED", True):
+            authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+        assert authed_session._credentials == self.credentials
+        await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_constructor_with_provided_auth_request(self):
+        auth_request = MockRequest()
+        authed_session = sessions.AsyncAuthorizedSession(
+            self.credentials, auth_request=auth_request
+        )
+
+        assert authed_session._auth_request is auth_request
+        await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_constructor_raises_no_auth_request_error(self):
+        with patch("google.auth.aio.transport.sessions.AIOHTTP_INSTALLED", False):
+            with pytest.raises(TransportError) as exc:
+                sessions.AsyncAuthorizedSession(self.credentials)
+
+        exc.match(
+            "`auth_request` must either be configured or the external package `aiohttp` must be installed to use the default value."
+        )
+
+    @pytest.mark.asyncio
+    async def test_constructor_raises_incorrect_credentials_error(self):
+        credentials = Mock()
+        with pytest.raises(InvalidType) as exc:
+            sessions.AsyncAuthorizedSession(credentials)
+
+        exc.match(
+            f"The configured credentials of type {type(credentials)} are invalid and must be of type `google.auth.aio.credentials.Credentials`"
+        )
+
+    @pytest.mark.asyncio
+    async def test_request_default_auth_request_success(self):
+        with aioresponses() as m:
+            mocked_chunks = [b"Cavefish ", b"have ", b"no ", b"sight."]
+            mocked_response = b"".join(mocked_chunks)
+            m.get(self.TEST_URL, status=200, body=mocked_response)
+            authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+            response = await authed_session.request("GET", self.TEST_URL)
+            assert response.status_code == 200
+            assert response.headers == {"Content-Type": "application/json"}
+            assert await response.read() == b"Cavefish have no sight."
+            await response.close()
+
+        await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_request_provided_auth_request_success(self, mocked_content):
+        mocked_response = MockResponse(
+            status_code=200,
+            headers={"Content-Type": "application/json"},
+            content=mocked_content,
+        )
+        auth_request = MockRequest(mocked_response)
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials, auth_request)
+        response = await authed_session.request("GET", self.TEST_URL)
+        assert response.status_code == 200
+        assert response.headers == {"Content-Type": "application/json"}
+        assert await response.read() == b"Cavefish have no sight."
+        await response.close()
+        assert response._close
+
+        await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_request_raises_timeout_error(self):
+        auth_request = MockRequest(side_effect=asyncio.TimeoutError)
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials, auth_request)
+        with pytest.raises(TimeoutError):
+            await authed_session.request("GET", self.TEST_URL)
+
+    @pytest.mark.asyncio
+    async def test_request_raises_transport_error(self):
+        auth_request = MockRequest(side_effect=TransportError)
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials, auth_request)
+        with pytest.raises(TransportError):
+            await authed_session.request("GET", self.TEST_URL)
+
+    @pytest.mark.asyncio
+    async def test_request_max_allowed_time_exceeded_error(self):
+        auth_request = MockRequest(side_effect=TransportError)
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials, auth_request)
+        with patch("time.monotonic", side_effect=[0, 1, 1]):
+            with pytest.raises(TimeoutError):
+                await authed_session.request("GET", self.TEST_URL, max_allowed_time=1)
+
+    @pytest.mark.parametrize("retry_status", DEFAULT_RETRYABLE_STATUS_CODES)
+    @pytest.mark.asyncio
+    async def test_request_max_retries(self, retry_status):
+        mocked_response = MockResponse(status_code=retry_status)
+        auth_request = MockRequest(mocked_response)
+        with patch("asyncio.sleep", return_value=None):
+            authed_session = sessions.AsyncAuthorizedSession(
+                self.credentials, auth_request
+            )
+            await authed_session.request("GET", self.TEST_URL)
+            assert auth_request.call_count == DEFAULT_MAX_RETRY_ATTEMPTS
+
+    @pytest.mark.asyncio
+    async def test_http_get_method_success(self):
+        expected_payload = b"content is retrieved."
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+        with aioresponses() as m:
+            m.get(self.TEST_URL, status=200, body=expected_payload)
+            response = await authed_session.get(self.TEST_URL)
+            assert await response.read() == expected_payload
+            response = await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_http_post_method_success(self):
+        expected_payload = b"content is posted."
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+        with aioresponses() as m:
+            m.post(self.TEST_URL, status=200, body=expected_payload)
+            response = await authed_session.post(self.TEST_URL)
+            assert await response.read() == expected_payload
+            response = await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_http_put_method_success(self):
+        expected_payload = b"content is retrieved."
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+        with aioresponses() as m:
+            m.put(self.TEST_URL, status=200, body=expected_payload)
+            response = await authed_session.put(self.TEST_URL)
+            assert await response.read() == expected_payload
+            response = await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_http_patch_method_success(self):
+        expected_payload = b"content is retrieved."
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+        with aioresponses() as m:
+            m.patch(self.TEST_URL, status=200, body=expected_payload)
+            response = await authed_session.patch(self.TEST_URL)
+            assert await response.read() == expected_payload
+            response = await authed_session.close()
+
+    @pytest.mark.asyncio
+    async def test_http_delete_method_success(self):
+        expected_payload = b"content is deleted."
+        authed_session = sessions.AsyncAuthorizedSession(self.credentials)
+        with aioresponses() as m:
+            m.delete(self.TEST_URL, status=200, body=expected_payload)
+            response = await authed_session.delete(self.TEST_URL)
+            assert await response.read() == expected_payload
+            response = await authed_session.close()
diff --git a/contrib/python/google-auth/py3/tests/ya.make b/contrib/python/google-auth/py3/tests/ya.make
index 6c6db898c4..23e821bb9a 100644
--- a/contrib/python/google-auth/py3/tests/ya.make
+++ b/contrib/python/google-auth/py3/tests/ya.make
@@ -9,6 +9,8 @@ PEERDIR(
     contrib/python/pytest-localserver
     contrib/python/oauth2client
     contrib/python/freezegun
+    contrib/python/aioresponses
+    contrib/python/pytest-asyncio
 )
 
 DATA(
@@ -22,16 +24,16 @@ PY_SRCS(
 )
 
 TEST_SRCS(
-    __init__.py
     compute_engine/__init__.py
-    compute_engine/test__metadata.py
     compute_engine/test_credentials.py
+    compute_engine/test__metadata.py
     conftest.py
     crypt/__init__.py
     crypt/test__cryptography_rsa.py
-    crypt/test__python_rsa.py
     crypt/test_crypt.py
     crypt/test_es256.py
+    crypt/test__python_rsa.py
+    __init__.py
     oauth2/__init__.py
     oauth2/test__client.py
     # oauth2/test_challenges.py - need pyu2f
@@ -42,35 +44,38 @@ TEST_SRCS(
     oauth2/test_service_account.py
     oauth2/test_sts.py
     oauth2/test_utils.py
-    oauth2/test_webauthn_handler.py
     oauth2/test_webauthn_handler_factory.py
+    oauth2/test_webauthn_handler.py
     oauth2/test_webauthn_types.py
-    test__cloud_sdk.py
-    test__default.py
-    test__exponential_backoff.py
-    test__helpers.py
-    test__oauth2client.py
-    test__refresh_worker.py
-    test__service_account_info.py
     test_api_key.py
     test_app_engine.py
     test_aws.py
+    test__cloud_sdk.py
+    test_credentials_async.py
     test_credentials.py
+    test__default.py
     test_downscoped.py
     test_exceptions.py
-    test_external_account.py
+    test__exponential_backoff.py
     test_external_account_authorized_user.py
+    test_external_account.py
+    test__helpers.py
     test_iam.py
     test_identity_pool.py
     test_impersonated_credentials.py
     test_jwt.py
     test_metrics.py
+    test__oauth2client.py
     test_packaging.py
     test_pluggable.py
+    test__refresh_worker.py
+    test__service_account_info.py
+    transport/aio/test_aiohttp.py
+    # transport/aio/test_sessions.py
     # transport/test__custom_tls_signer.py
+    transport/test_grpc.py
     transport/test__http_client.py
     transport/test__mtls_helper.py
-    transport/test_grpc.py
     transport/test_mtls.py
     # transport/test_requests.py
     # transport/test_urllib3.py
diff --git a/contrib/python/google-auth/py3/ya.make b/contrib/python/google-auth/py3/ya.make
index 4f5c4e4ad8..60146f91a4 100644
--- a/contrib/python/google-auth/py3/ya.make
+++ b/contrib/python/google-auth/py3/ya.make
@@ -2,7 +2,7 @@
 
 PY3_LIBRARY()
 
-VERSION(2.34.0)
+VERSION(2.35.0)
 
 LICENSE(Apache-2.0)
 
@@ -20,6 +20,7 @@ NO_LINT()
 
 NO_CHECK_IMPORTS(
     google.auth._oauth2client
+    google.auth.aio.transport.aiohttp
     google.auth.transport._aiohttp_requests
 )
 
@@ -39,6 +40,9 @@ PY_SRCS(
     google/auth/_service_account_info.py
     google/auth/aio/__init__.py
     google/auth/aio/credentials.py
+    google/auth/aio/transport/__init__.py
+    google/auth/aio/transport/aiohttp.py
+    google/auth/aio/transport/sessions.py
     google/auth/api_key.py
     google/auth/app_engine.py
     google/auth/aws.py