googleapis
diff --git a/‎packages/google-auth/google/auth/_regional_access_boundary_utils.py‎
Lines changed: 85 additions & 3 deletions b/‎packages/google-auth/google/auth/_regional_access_boundary_utils.py‎
Lines changed: 85 additions & 3 deletions
diff --git a/‎packages/google-auth/google/auth/aio/transport/__init__.py‎
Lines changed: 10 additions & 0 deletions b/‎packages/google-auth/google/auth/aio/transport/__init__.py‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎packages/google-auth/google/auth/aio/transport/aiohttp.py‎
Lines changed: 81 additions & 1 deletion b/‎packages/google-auth/google/auth/aio/transport/aiohttp.py‎
Lines changed: 81 additions & 1 deletion
diff --git a/‎packages/google-auth/google/auth/transport/_aiohttp_requests.py‎
Lines changed: 90 additions & 0 deletions b/‎packages/google-auth/google/auth/transport/_aiohttp_requests.py‎
Lines changed: 90 additions & 0 deletions
@@ -27,7 +27,7 @@
 from google.auth import _helpers
 from google.auth import environment_vars
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: NO COVER
     import google.auth.credentials
     import google.auth.transport
 
@@ -455,6 +455,61 @@ def start_refresh(self, credentials, request, rab_manager):
             self._worker.start()
 
 
+def _prepare_async_lookup_callable(request):
+    """Unwraps a request callable, clones the transport, and returns the new callable.
+
+    Args:
+        request: The original request callable (e.g. functools.partial or raw Request).
+
+    Returns:
+        Tuple[Callable, Any, bool]: A tuple containing the new lookup callable, the
+            underlying request object, and a boolean indicating if it was cloned.
+    """
+    is_partial = isinstance(request, functools.partial)
+    base_callable = request.func if is_partial else request
+
+    if not hasattr(base_callable, "_clone"):
+        return request, base_callable, False
+
+    cloned_callable = base_callable._clone()
+    is_cloned = cloned_callable is not base_callable
+
+    if is_partial:
+        new_request = functools.partial(
+            cloned_callable, *request.args, **request.keywords
+        )
+    else:
+        new_request = cloned_callable
+
+    return new_request, cloned_callable, is_cloned
+
+
+async def _close_cloned_request(lookup_request, is_cloned):
+    """Safely closes the underlying cloned request transport, if applicable.
+
+    Args:
+        lookup_request (Any): The request object/transport to close.
+        is_cloned (bool): Whether the request was actually cloned.
+    """
+    if not is_cloned or not hasattr(lookup_request, "close"):
+        return
+
+    is_async = False
+    try:
+        maybe_coro = lookup_request.close()
+        if is_async := inspect.isawaitable(maybe_coro):
+            await maybe_coro
+    except Exception as e:
+        if _helpers.is_logging_enabled(_LOGGER):
+            adapter_type = " asynchronous " if is_async else " "
+            _LOGGER.warning(
+                "Failed to cleanly close cloned%srequest transport: %s",
+                adapter_type,
+                e,
+                exc_info=True,
+            )
+
+
 class _AsyncRegionalAccessBoundaryRefreshManager(object):
     """Manages a task for background refreshing of the Regional Access Boundary in async flows."""
 
@@ -491,11 +546,28 @@ def start_refresh(self, credentials, request, rab_manager):
                 # A refresh is already in progress.
                 return
 
+            try:
+                (
+                    lookup_callable,
+                    lookup_request,
+                    is_cloned,
+                ) = _prepare_async_lookup_callable(request)
+            except Exception as e:
+                if _helpers.is_logging_enabled(_LOGGER):
+                    _LOGGER.warning(
+                        "Synchronous cloning of request for Regional Access Boundary lookup failed: %s",
+                        e,
+                        exc_info=True,
+                    )
+                rab_manager.process_regional_access_boundary_info(None)
+                return
+
             async def _worker():
                 try:
-                    # credentials._lookup_regional_access_boundary should be async in the async creds class
                     regional_access_boundary_info = (
-                        await credentials._lookup_regional_access_boundary(request)
+                        await credentials._lookup_regional_access_boundary(
+                            lookup_callable
+                        )
                     )
                 except Exception as e:
                     if _helpers.is_logging_enabled(_LOGGER):
@@ -505,6 +577,8 @@ async def _worker():
                             exc_info=True,
                         )
                     regional_access_boundary_info = None
+                finally:
+                    await _close_cloned_request(lookup_request, is_cloned)
 
                 rab_manager.process_regional_access_boundary_info(
                     regional_access_boundary_info
@@ -514,7 +588,15 @@ async def _worker():
             try:
                 self._worker_task = asyncio.create_task(coro)
             except Exception:
+                # Clean up cloned request if task creation fails
                 coro.close()
+                try:
+                    asyncio.get_running_loop().create_task(
+                        _close_cloned_request(lookup_request, is_cloned)
+                    )
+                except RuntimeError:
+                    pass
+                rab_manager.process_regional_access_boundary_info(None)
                 raise
 
 
 
@@ -142,3 +142,13 @@ async def close(self) -> None:
         Close the underlying session.
         """
         raise NotImplementedError("close must be implemented.")
+
+    def _clone(self) -> "Request":
+        """Creates a copy of this request adapter.
+
+        The base implementation returns `self` (an identical shared instance).
+        Transport adapters that maintain internal connection pools or stateful
+        sessions must override this method to return an independent, detached
+        adapter instance.
+        """
+        return self
@@ -36,7 +36,7 @@
 else:
     try:
         from aiohttp import ClientTimeout
-    except (ImportError, AttributeError):
+    except (ImportError, AttributeError):  # pragma: NO COVER
         ClientTimeout = None
 
 _LOGGER = logging.getLogger(__name__)
@@ -203,3 +203,83 @@ async def close(self) -> None:
         if not self._closed and self._session:
             await self._session.close()
         self._closed = True
+
+    def _clone(self) -> "Request":
+        """Creates an independent copy of this request adapter.
+
+        Clones the connection settings, trace configurations, and session defaults
+        (headers, cookies, basic auth, and timeouts).
+
+        Only standard `aiohttp.TCPConnector` and `aiohttp.UnixConnector` connectors
+        are supported. The DNS resolver is not copied to avoid closing shared resolver
+        resources.
+
+        Returns:
+            google.auth.aio.transport.aiohttp.Request: A new request adapter.
+
+        Raises:
+            google.auth.exceptions.TransportError: If the transport is closed, or if the
+                session uses an unsupported connector.
+        """
+        if self._closed:
+            raise exceptions.TransportError("Cannot clone a closed transport.")
+
+        if not self._session:
+            new_session = aiohttp.ClientSession(
+                auto_decompress=False,
+                trust_env=True,
+            )
+            return Request(session=new_session)
+
+        session_kwargs: dict = {
+            "auto_decompress": False,
+            "trust_env": getattr(self._session, "_trust_env", True),
+        }
+
+        # Copy underlying connection pool settings (SSL context, IP bindings, limits).
+        orig_connector = getattr(self._session, "_connector", None)
+        if orig_connector and not orig_connector.closed:
+            if isinstance(orig_connector, aiohttp.TCPConnector):
+                # We explicitly do not copy the resolver. The connector
+                # owns the resolver, and closing the cloned session would
+                # close the shared resolver, breaking the original session.
+                session_kwargs["connector"] = aiohttp.TCPConnector(
+                    ssl=getattr(orig_connector, "_ssl", None),  # type: ignore
+                    limit=getattr(orig_connector, "_limit", 100),
+                    limit_per_host=getattr(orig_connector, "_limit_per_host", 0),
+                    force_close=getattr(orig_connector, "_force_close", False),
+                    local_addr=getattr(orig_connector, "_local_addr", None),
+                )
+            elif getattr(aiohttp, "UnixConnector", None) and isinstance(
+                orig_connector, getattr(aiohttp, "UnixConnector")
+            ):
+                path = getattr(orig_connector, "_path", None)
+                if path:
+                    session_kwargs["connector"] = aiohttp.UnixConnector(
+                        path=path,
+                        limit=getattr(orig_connector, "_limit", 100),
+                        force_close=getattr(orig_connector, "_force_close", False),
+                    )
+            else:
+                raise exceptions.TransportError(
+                    f"Unsupported connector type for cloning: {type(orig_connector)}"
+                )
+
+        # Preserve distributed tracing configurations.
+        trace_configs = getattr(self._session, "_trace_configs", None)
+        if trace_configs:
+            session_kwargs["trace_configs"] = list(trace_configs)
+
+        # Copy session-level defaults (headers, cookies, auth, timeout).
+        for attr_name, kwarg_name in [
+            ("_default_headers", "headers"),
+            ("_cookie_jar", "cookie_jar"),
+            ("_default_auth", "auth"),
+            ("_timeout", "timeout"),
+            ("_json_serialize", "json_serialize"),
+        ]:
+            val = getattr(self._session, attr_name, None)
+            if val is not None:
+                session_kwargs[kwarg_name] = val
+
+        return Request(session=aiohttp.ClientSession(**session_kwargs))  # type: ignore
@@ -148,6 +148,7 @@ def __init__(self, session=None):
                 "Client sessions with auto_decompress=True are not supported."
             )
         self.session = session
+        self._closed = False
 
     async def __call__(
         self,
@@ -183,6 +184,9 @@ async def __call__(
         """
 
         try:
+            if getattr(self, "_closed", False):
+                raise exceptions.TransportError("session is closed.")
+
             if self.session is None:  # pragma: NO COVER
                 self.session = aiohttp.ClientSession(
                     auto_decompress=False
@@ -202,6 +206,92 @@ async def __call__(
             new_exc = exceptions.TransportError(caught_exc)
             raise new_exc from caught_exc
 
+    def _clone(self):
+        """Creates an independent copy of this request adapter.
+
+        Clones the connection settings, trace configurations, and session defaults
+        (headers, cookies, basic auth, and timeouts).
+
+        Only standard `aiohttp.TCPConnector` and `aiohttp.UnixConnector` connectors
+        are supported. The DNS resolver is not copied to avoid closing shared resolver
+        resources.
+
+        Returns:
+            google.auth.transport._aiohttp_requests.Request: A new request adapter.
+
+        Raises:
+            google.auth.exceptions.TransportError: If the transport is closed, or if the
+                session uses an unsupported connector.
+        """
+        if getattr(self, "_closed", False):
+            raise exceptions.TransportError("Cannot clone a closed transport.")
+
+        if not self.session:
+            new_session = aiohttp.ClientSession(
+                auto_decompress=False,
+                trust_env=True,
+            )
+            return Request(session=new_session)
+
+        session_kwargs: dict = {
+            "auto_decompress": False,
+            "trust_env": getattr(self.session, "_trust_env", True),
+        }
+
+        # Copy underlying connection pool settings (SSL context, IP bindings, limits).
+        orig_connector = getattr(self.session, "_connector", None)
+        if orig_connector and not getattr(orig_connector, "closed", True):
+            if isinstance(orig_connector, aiohttp.TCPConnector):
+                # We explicitly do not copy the resolver. The connector
+                # owns the resolver, and closing the cloned session would
+                # close the shared resolver, breaking the original session.
+                session_kwargs["connector"] = aiohttp.TCPConnector(
+                    ssl=getattr(orig_connector, "_ssl", None),  # type: ignore
+                    limit=getattr(orig_connector, "_limit", 100),
+                    limit_per_host=getattr(orig_connector, "_limit_per_host", 0),
+                    force_close=getattr(orig_connector, "_force_close", False),
+                    local_addr=getattr(orig_connector, "_local_addr", None),
+                )
+            elif getattr(aiohttp, "UnixConnector", None) and isinstance(
+                orig_connector, getattr(aiohttp, "UnixConnector")
+            ):
+                path = getattr(orig_connector, "_path", None)
+                if path:
+                    session_kwargs["connector"] = aiohttp.UnixConnector(
+                        path=path,
+                        limit=getattr(orig_connector, "_limit", 100),
+                        force_close=getattr(orig_connector, "_force_close", False),
+                    )
+            else:
+                raise exceptions.TransportError(
+                    f"Unsupported connector type for cloning: {type(orig_connector)}"
+                )
+
+        # Preserve distributed tracing configurations.
+        trace_configs = getattr(self.session, "_trace_configs", None)
+        if trace_configs:
+            session_kwargs["trace_configs"] = list(trace_configs)
+
+        # Copy session-level defaults (headers, cookies, auth, timeout).
+        for attr_name, kwarg_name in [
+            ("_default_headers", "headers"),
+            ("_cookie_jar", "cookie_jar"),
+            ("_default_auth", "auth"),
+            ("_timeout", "timeout"),
+            ("_json_serialize", "json_serialize"),
+        ]:
+            val = getattr(self.session, attr_name, None)
+            if val is not None:
+                session_kwargs[kwarg_name] = val
+
+        return Request(session=aiohttp.ClientSession(**session_kwargs))  # type: ignore
+
+    async def close(self):
+        """Cleanly release the underlying aiohttp ClientSession resources."""
+        if not getattr(self, "_closed", False) and self.session:
+            await self.session.close()
+        self._closed = True
+
 
 class AuthorizedSession(aiohttp.ClientSession):
     """This is an async implementation of the Authorized Session class. We utilize an