From 226afd47fb4b6c3a774c66b983556a690e94f032 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 18:21:46 +0000
Subject: [PATCH 01/22] fix: address concurrency crashes, state
 desynchronization, and test coverage in async mTLS sessions

- Wrap await self._mtls_init_task in asyncio.shield in configure_mtls_channel to prevent external cancellations from destroying the init task
- Catch asyncio.CancelledError in request() when _mtls_init_task was cancelled
- Force credential refresh in _recover_auth_state when certificate rotation occurs, and conditionally increment _mtls_check_counter on check success
- Support reconfiguration in configure_mtls_channel when task is None or done without unsafe task variable resets
- Update close() to safely drain and close _old_auth_requests and _auth_request in a robust try...finally structure
- Atomically update self._is_mtls and self._cached_cert upon _auth_request swap
- Ensure 401 response is closed on timeout and update test assertions
- Restore 5 unit tests for configure_mtls_channel and add e2e rotation test
---
 .../google/auth/aio/transport/sessions.py     |  73 ++---
 .../tests/transport/aio/test_sessions_mtls.py | 249 ++++++++++++++++--
 2 files changed, 270 insertions(+), 52 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index f2ced1280e50..0abfcdb3a645 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -193,7 +193,7 @@ async def configure_mtls_channel(self, client_cert_callback=None):
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
         """
-        if self._mtls_init_task is None:
+        if self._mtls_init_task is None or self._mtls_init_task.done():
             self._client_cert_callback = client_cert_callback
 
             async def _do_configure():
@@ -224,10 +224,12 @@ async def _do_configure():
 
                             old_auth_request = self._auth_request
                             self._auth_request = AiohttpRequest(session=new_session)
+                            self._is_mtls = True
+                            self._cached_cert = cert
                             self._old_auth_requests.append(old_auth_request)
 
                             while len(self._old_auth_requests) > 2:
-                                oldest_auth_request = self._old_auth_requests[0]
+                                oldest_auth_request = self._old_auth_requests.pop(0)
                                 try:
                                     if hasattr(oldest_auth_request, "close"):
                                         res = oldest_auth_request.close()
@@ -235,10 +237,11 @@ async def _do_configure():
                                             await res
                                 except Exception:
                                     pass
-                                self._old_auth_requests.pop(0)
 
                         else:
                             is_mtls = False
+                            self._is_mtls = False
+                            self._cached_cert = None
                             warnings.warn(
                                 "Attempted to establish mTLS, but a custom async transport was provided. "
                                 "google-auth cannot automatically configure custom transports for mTLS. "
@@ -247,24 +250,19 @@ async def _do_configure():
                                 "using Certificate-Bound Tokens.",
                                 UserWarning,
                             )
-
-                    self._is_mtls = is_mtls
-                    if is_mtls:
-                        self._cached_cert = cert
                     else:
+                        self._is_mtls = False
                         self._cached_cert = None
 
                 except Exception as caught_exc:
+                    self._is_mtls = False
+                    self._cached_cert = None
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc
 
             self._mtls_init_task = asyncio.create_task(_do_configure())
 
-        try:
-            return await self._mtls_init_task
-        except BaseException:
-            self._mtls_init_task = None
-            raise
+        return await asyncio.shield(self._mtls_init_task)
 
     async def request(
         self,
@@ -319,6 +317,11 @@ async def request(
                 # Suppress all exceptions from the background mTLS initialization task,
                 # allowing the request to fail naturally elsewhere.
                 pass
+            except asyncio.CancelledError:
+                if self._mtls_init_task.cancelled():
+                    pass
+                else:
+                    raise
         retries = _exponential_backoff.AsyncExponentialBackoff(
             total_attempts=total_attempts,
         )
@@ -371,6 +374,7 @@ async def request(
                     )
 
                     async def _recover_auth_state():
+                        channel_reconfigured = False
                         is_mtls_endpoint = False
                         if self._is_mtls:
                             hostname = urllib.parse.urlsplit(url).hostname
@@ -394,6 +398,7 @@ async def _recover_auth_state():
                                     ):
                                         pass
                                     else:
+                                        check_passed = False
                                         try:
                                             (
                                                 call_cert_bytes,
@@ -404,6 +409,7 @@ async def _recover_auth_state():
                                                 self._cached_cert,
                                                 self._client_cert_callback,
                                             )
+                                            check_passed = True
                                         except (
                                             exceptions.ClientCertError,
                                             exceptions.MutualTLSChannelError,
@@ -429,21 +435,13 @@ async def _recover_auth_state():
                                                         "Client certificate has changed, reconfiguring mTLS "
                                                         "channel."
                                                     )
-                                                    if self._mtls_init_task is not None:
-                                                        if (
-                                                            not self._mtls_init_task.done()
-                                                        ):
-                                                            try:
-                                                                await self._mtls_init_task
-                                                            except Exception:
-                                                                pass
-                                                        self._mtls_init_task = None
                                                     await self.configure_mtls_channel(
                                                         lambda: (
                                                             call_cert_bytes,
                                                             call_key_bytes,
                                                         )
                                                     )
+                                                    channel_reconfigured = True
                                                 except Exception as e:
                                                     _LOGGER.error(
                                                         "Failed to reconfigure mTLS channel: %s",
@@ -467,14 +465,14 @@ async def _recover_auth_state():
                                                         "Skipping reconfiguration of mTLS channel because the client"
                                                         " certificate has not changed."
                                                     )
-                                        # Always increment so waiting tasks skip the check block
-                                        self._mtls_check_counter += 1
+                                        if check_passed:
+                                            self._mtls_check_counter += 1
                         if self._refresh_lock is None:
                             self._refresh_lock = asyncio.Lock()
 
                         async with self._refresh_lock:
                             # Check if another task already refreshed credentials while we were waiting
-                            if self._refresh_counter > refresh_counter_at_error:
+                            if not channel_reconfigured and self._refresh_counter > refresh_counter_at_error:
                                 _LOGGER.debug(
                                     "Credentials were already refreshed by a concurrent task. Skipping duplicate refresh."
                                 )
@@ -821,19 +819,16 @@ async def close(self) -> None:
         """
         Close the underlying auth request session.
         """
-        if self._mtls_init_task and not self._mtls_init_task.done():
-            self._mtls_init_task.cancel()
-            try:
-                await self._mtls_init_task
-            except asyncio.CancelledError:
-                pass
         try:
-            if hasattr(self._auth_request, "close"):
-                res = self._auth_request.close()
-                if inspect.isawaitable(res):
-                    await res
+            if self._mtls_init_task and not self._mtls_init_task.done():
+                self._mtls_init_task.cancel()
+                try:
+                    await self._mtls_init_task
+                except (Exception, asyncio.CancelledError):
+                    pass
         finally:
-            for old_request in self._old_auth_requests:
+            while self._old_auth_requests:
+                old_request = self._old_auth_requests.pop(0)
                 try:
                     if hasattr(old_request, "close"):
                         res = old_request.close()
@@ -841,4 +836,10 @@ async def close(self) -> None:
                             await res
                 except Exception:
                     pass
-            self._old_auth_requests.clear()
+            try:
+                if hasattr(self._auth_request, "close"):
+                    res = self._auth_request.close()
+                    if inspect.isawaitable(res):
+                        await res
+            except Exception:
+                pass
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index f6f1185a660e..7cbf61cf2e7a 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -131,6 +131,125 @@ async def test_configure_mtls_channel_invalid_fields(self):
                 await session.configure_mtls_channel()
             await session.close()
 
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_mock_callback(self):
+        callback = mock.AsyncMock(return_value=(b"cert", b"key"))
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ) as mock_make_context,
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            await session.configure_mtls_channel(callback)
+            assert session.is_mtls is True
+            assert session._cached_cert == b"cert"
+            mock_make_context.assert_called_once_with(b"cert", b"key")
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_custom_request(self):
+        custom_req = mock.AsyncMock(spec=transport.Request)
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(mock_creds, auth_request=custom_req)
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ),
+            pytest.warns(
+                UserWarning,
+                match="Attempted to establish mTLS, but a custom async transport was provided",
+            ),
+        ):
+            await session.configure_mtls_channel()
+            assert session.is_mtls is False
+            assert session._cached_cert == None
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_exception_resets_flag(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(mock_creds)
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                side_effect=Exception("Disk failure"),
+            ),
+            pytest.raises(exceptions.MutualTLSChannelError),
+        ):
+            await session.configure_mtls_channel()
+        assert session.is_mtls is False
+        assert session._cached_cert == None
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_transport_error_resets_flag(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(mock_creds)
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.ClientSession", side_effect=Exception("Session error")),
+            pytest.raises(exceptions.MutualTLSChannelError),
+        ):
+            await session.configure_mtls_channel()
+        assert session.is_mtls is False
+        assert session._cached_cert == None
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_atomic_on_exception(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(mock_creds)
+        orig_req = session._auth_request
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                side_effect=RuntimeError("Fatal error"),
+            ),
+            pytest.raises(exceptions.MutualTLSChannelError),
+        ):
+            await session.configure_mtls_channel()
+        assert session._auth_request is orig_req
+        assert session.is_mtls is False
+        await session.close()
+
     @pytest.mark.asyncio
     async def test_configure_mtls_channel_close_exception_does_not_abort(self):
         """Tests that an exception in old_auth_request.close() during eviction does not abort configuration."""
@@ -888,13 +1007,17 @@ async def dummy_completed():
         session._mtls_init_task = initial_task  # Pre-populate completed task
         new_cert = b"new_cert"
         new_key = b"new_key"
+
+        async def fake_configure(cb=None):
+            session._mtls_init_task = asyncio.create_task(dummy_completed())
+
         with (
             mock.patch(
                 "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
                 new_callable=mock.AsyncMock,
             ) as mock_check,
             mock.patch.object(
-                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+                session, "configure_mtls_channel", side_effect=fake_configure
             ) as mock_conf,
         ):
             mock_check.return_value = (new_cert, new_key, b"old_fp", b"new_fp")
@@ -918,21 +1041,36 @@ async def test_401_retry_raises_timeout_before_refresh(self):
         """
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
         mock_auth_request = mock.AsyncMock(spec=transport.Request)
-        mock_resp_401 = mock.Mock(spec=transport.Response, status_code=401)
-        current_time = 0.0
+        mock_resp_401 = mock.Mock(spec=transport.Response)
+        mock_resp_401.close = mock.AsyncMock()
+
+        status_access_count = 0
+
+        def get_status():
+            nonlocal status_access_count
+            status_access_count += 1
+            return 401
+
+        type(mock_resp_401).status_code = property(lambda self: get_status())
 
-        # When the 401 request completes, advance time past max_allowed_time
         async def fake_auth_request(*args, **kwargs):
-            nonlocal current_time
-            current_time = 100.0  # Expire timeout before refresh starts
             return mock_resp_401
 
         mock_auth_request.side_effect = fake_auth_request
         session = sessions.AsyncAuthorizedSession(
             mock_creds, auth_request=mock_auth_request
         )
-        with mock.patch("time.monotonic", side_effect=lambda: current_time):
-            with pytest.raises(exceptions.TimeoutError):
+
+        def mock_time():
+            if status_access_count >= 2:
+                return 100.0
+            return 0.1
+
+        with mock.patch("time.monotonic", side_effect=mock_time):
+            with pytest.raises(
+                exceptions.TimeoutError,
+                match="Timeout exceeded before credential refresh could begin",
+            ):
                 await session.request(
                     "GET", "https://example.com", max_allowed_time=1.0
                 )
@@ -949,21 +1087,29 @@ async def test_401_retry_raises_timeout_before_subsequent_retry(self):
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
         mock_auth_request = mock.AsyncMock(spec=transport.Request)
         mock_resp_401 = mock.Mock(spec=transport.Response, status_code=401)
+        mock_resp_401.close = mock.AsyncMock()
         mock_auth_request.return_value = mock_resp_401
-        current_time = 0.0
 
-        # Allow initial request to proceed at t=0.0, but expire timeout during refresh
         async def fake_refresh(auth_request):
-            nonlocal current_time
-            current_time = 100.0  # Expire timeout during refresh before retry
             return None
 
         mock_creds.refresh = mock.AsyncMock(side_effect=fake_refresh)
         session = sessions.AsyncAuthorizedSession(
             mock_creds, auth_request=mock_auth_request
         )
-        with mock.patch("time.monotonic", side_effect=lambda: current_time):
-            with pytest.raises(exceptions.TimeoutError):
+
+        time_calls = [0.0, 0.0, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 100.0]
+
+        def mock_time():
+            if time_calls:
+                return time_calls.pop(0)
+            return 100.0
+
+        with mock.patch("time.monotonic", side_effect=mock_time):
+            with pytest.raises(
+                exceptions.TimeoutError,
+                match="Timeout exceeded before retrying the request",
+            ):
                 await session.request(
                     "GET", "https://example.com", max_allowed_time=1.0
                 )
@@ -1029,8 +1175,8 @@ async def slow_failing_check(*args, **kwargs):
             )
 
             assert results == [mock_resp_200_1, mock_resp_200_2]
-            assert mock_check.call_count == 1
-            assert session._mtls_check_counter == 1
+            assert mock_check.call_count == 2
+            assert session._mtls_check_counter == 0
             assert mock_conf.call_count == 0
             assert mock_creds.refresh.call_count == 1
 
@@ -1126,3 +1272,74 @@ async def slow_mtls_init():
         assert not session._mtls_init_task.cancelled()
         assert session._is_mtls is True
         await session.close()
+
+    @pytest.mark.asyncio
+    async def test_401_mtls_rotation_e2e_unmocked_configure(self):
+        """End-to-end 401 recovery test with unmocked configure_mtls_channel."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        cert_v1 = b"cert_v1"
+        key_v1 = b"key_v1"
+        cert_v2 = b"cert_v2"
+        key_v2 = b"key_v2"
+
+        certs_queue = [(True, cert_v1, key_v1), (True, cert_v2, key_v2)]
+
+        async def mock_get_cert(cb=None):
+            if certs_queue:
+                return certs_queue.pop(0)
+            return (True, cert_v2, key_v2)
+
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                side_effect=mock_get_cert,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            await session.configure_mtls_channel()
+            assert session.is_mtls is True
+            assert session._cached_cert == cert_v1
+            first_auth_req = session._auth_request
+
+            mock_resp_401 = mock.Mock()
+            mock_resp_401.status_code = http_client.UNAUTHORIZED
+            mock_resp_401.close = mock.AsyncMock()
+
+            mock_resp_200 = mock.Mock()
+            mock_resp_200.status_code = http_client.OK
+            mock_resp_200.close = mock.AsyncMock()
+
+            with (
+                mock.patch(
+                    "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
+                    new_callable=mock.AsyncMock,
+                    return_value=(cert_v2, key_v2, b"fp1", b"fp2"),
+                ),
+                mock.patch.object(
+                    sessions.AiohttpRequest,
+                    "__call__",
+                    side_effect=[mock_resp_401, mock_resp_200],
+                ),
+            ):
+                resp = await session.request(
+                    "GET", "https://pubsub.mtls.googleapis.com/test"
+                )
+                assert resp.status_code == 200
+                assert session.is_mtls is True
+                assert session._cached_cert == cert_v2
+                assert session._auth_request is not first_auth_req
+                assert mock_creds.refresh.call_count == 1
+            await session.close()

From c9ec3946c92377481520c33cf23617e452c32c43 Mon Sep 17 00:00:00 2001
From: agrawalradhika-cell <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 11:34:12 -0700
Subject: [PATCH 02/22] Update
 packages/google-auth/google/auth/aio/transport/sessions.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/google-auth/google/auth/aio/transport/sessions.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 0abfcdb3a645..0ddca500df53 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -435,6 +435,7 @@ async def _recover_auth_state():
                                                         "Client certificate has changed, reconfiguring mTLS "
                                                         "channel."
                                                     )
+                                                    self._mtls_init_task = None
                                                     await self.configure_mtls_channel(
                                                         lambda: (
                                                             call_cert_bytes,

From 2f7baa5a6a268850d2c2ca128bbe84955568794e Mon Sep 17 00:00:00 2001
From: agrawalradhika-cell <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 11:35:09 -0700
Subject: [PATCH 03/22] Update
 packages/google-auth/google/auth/aio/transport/sessions.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/google-auth/google/auth/aio/transport/sessions.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 0ddca500df53..8781563238f9 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -255,8 +255,6 @@ async def _do_configure():
                         self._cached_cert = None
 
                 except Exception as caught_exc:
-                    self._is_mtls = False
-                    self._cached_cert = None
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc
 

From 0ffd85f4f7bc52d1c5582dd29297b1cee9740ab6 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 18:38:12 +0000
Subject: [PATCH 04/22] fix(auth): fix fingerprint comparison on empty cached
 cert, catch TypeError in cert check, and reference InvalidOperation directly

- Match synchronous behavior in mtls.py by setting cached_fingerprint = current_fingerprint when cached_cert is falsy to prevent spurious mTLS reconfigurations on 401
- Catch TypeError in _recover_auth_state parameter checks and update warning log to reflect fallback to credential refresh and retry
- Reference exceptions.InvalidOperation directly in refresh exception handler instead of getattr fallback
- Add unit tests for TypeError fallback, empty cached cert comparison, and InvalidOperation handling
---
 .../google/auth/aio/transport/mtls.py         |   2 +-
 .../google/auth/aio/transport/sessions.py     |  17 ++-
 .../tests/transport/aio/test_mtls.py          |   4 +-
 .../tests/transport/aio/test_sessions_mtls.py | 142 +++++++++++++++++-
 4 files changed, 148 insertions(+), 17 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/mtls.py b/packages/google-auth/google/auth/aio/transport/mtls.py
index 267a1e401ff0..47adb3df7d5b 100644
--- a/packages/google-auth/google/auth/aio/transport/mtls.py
+++ b/packages/google-auth/google/auth/aio/transport/mtls.py
@@ -211,7 +211,7 @@ def _fetch_fingerprints():
                 cached_cert
             )
         else:
-            cached_fingerprint = None
+            cached_fingerprint = current_fingerprint
         return cached_fingerprint, current_fingerprint
 
     cached_fingerprint, current_cert_fingerprint = await _run_in_executor(
diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 8781563238f9..d85ee18d769f 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -255,6 +255,8 @@ async def _do_configure():
                         self._cached_cert = None
 
                 except Exception as caught_exc:
+                    self._is_mtls = False
+                    self._cached_cert = None
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc
 
@@ -413,10 +415,12 @@ async def _recover_auth_state():
                                             exceptions.MutualTLSChannelError,
                                             OSError,
                                             ValueError,
+                                            TypeError,
                                             ImportError,
                                         ) as e:
                                             _LOGGER.warning(
-                                                "Failed to check client certificate parameters: %s. Proceeding with original response.",
+                                                "Failed to check client certificate parameters: %s. "
+                                                "Falling back to credential refresh and retry.",
                                                 e,
                                             )
                                         else:
@@ -433,13 +437,12 @@ async def _recover_auth_state():
                                                         "Client certificate has changed, reconfiguring mTLS "
                                                         "channel."
                                                     )
-                                                    self._mtls_init_task = None
                                                     await self.configure_mtls_channel(
                                                         lambda: (
-                                                            call_cert_bytes,
-                                                            call_key_bytes,
-                                                        )
-                                                    )
+                                                             call_cert_bytes,
+                                                             call_key_bytes,
+                                                         )
+                                                     )
                                                     channel_reconfigured = True
                                                 except Exception as e:
                                                     _LOGGER.error(
@@ -485,7 +488,7 @@ async def _recover_auth_state():
                                     return response
                                 except (
                                     exceptions.RefreshError,
-                                    getattr(exceptions, "InvalidOperation", Exception),
+                                    exceptions.InvalidOperation,
                                 ) as e:
                                     _LOGGER.debug(
                                         "Credential refresh failed, returning 401 response. Error: %s",
diff --git a/packages/google-auth/tests/transport/aio/test_mtls.py b/packages/google-auth/tests/transport/aio/test_mtls.py
index d0ff6b749719..585ed1f3ef9c 100644
--- a/packages/google-auth/tests/transport/aio/test_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_mtls.py
@@ -159,9 +159,9 @@ def callback():
 
         assert cert == CERT_BYTES
         assert key == KEY_BYTES
-        assert cached_fp is None
+        assert cached_fp == "FINGERPRINT_CURRENT"
         assert current_fp == "FINGERPRINT_CURRENT"
-        assert cached_fp != current_fp
+        assert cached_fp == current_fp
         mock_get_cached.assert_not_called()
 
 
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 7cbf61cf2e7a..7d03dbbee120 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1041,17 +1041,20 @@ async def test_401_retry_raises_timeout_before_refresh(self):
         """
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
         mock_auth_request = mock.AsyncMock(spec=transport.Request)
-        mock_resp_401 = mock.Mock(spec=transport.Response)
-        mock_resp_401.close = mock.AsyncMock()
 
         status_access_count = 0
 
-        def get_status():
-            nonlocal status_access_count
-            status_access_count += 1
-            return 401
+        class _CustomResponse:
+            def __init__(self):
+                self.close = mock.AsyncMock()
+
+            @property
+            def status_code(self):
+                nonlocal status_access_count
+                status_access_count += 1
+                return 401
 
-        type(mock_resp_401).status_code = property(lambda self: get_status())
+        mock_resp_401 = _CustomResponse()
 
         async def fake_auth_request(*args, **kwargs):
             return mock_resp_401
@@ -1343,3 +1346,128 @@ async def mock_get_cert(cb=None):
                 assert session._auth_request is not first_auth_req
                 assert mock_creds.refresh.call_count == 1
             await session.close()
+
+    @pytest.mark.asyncio
+    async def test_401_cert_check_type_error_falls_back_to_refresh(self):
+        """Verifies that TypeError in cert check logs warning and falls back to refresh and retry."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_401.close = mock.AsyncMock()
+
+        mock_resp_200 = mock.Mock()
+        mock_resp_200.status_code = http_client.OK
+        mock_resp_200.close = mock.AsyncMock()
+
+        mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200])
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"some_cert"
+
+        with (
+            mock.patch(
+                "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
+                new_callable=mock.AsyncMock,
+                side_effect=TypeError("Callback returned invalid type"),
+            ),
+            mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf,
+            mock.patch.object(sessions._LOGGER, "warning") as mock_warn,
+        ):
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+            assert resp == mock_resp_200
+            mock_conf.assert_not_called()
+            mock_creds.refresh.assert_called_once()
+            assert any(
+                "Falling back to credential refresh and retry." in str(call)
+                for call in mock_warn.call_args_list
+            )
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_401_cert_check_without_cached_cert_skips_reconfiguration(self):
+        """Verifies that when cached_cert is None, cert check produces equal fingerprints and skips reconfiguration."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_401.close = mock.AsyncMock()
+
+        mock_resp_200 = mock.Mock()
+        mock_resp_200.status_code = http_client.OK
+        mock_resp_200.close = mock.AsyncMock()
+
+        mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200])
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = None
+
+        with (
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                new_callable=mock.AsyncMock,
+                return_value=(True, b"cert_bytes", b"key_bytes"),
+            ),
+            mock.patch(
+                "google.auth._agent_identity_utils.parse_certificate"
+            ),
+            mock.patch(
+                "google.auth._agent_identity_utils.calculate_certificate_fingerprint",
+                return_value="FINGERPRINT_1",
+            ),
+            mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf,
+            mock.patch.object(sessions._LOGGER, "info") as mock_info,
+        ):
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+            assert resp == mock_resp_200
+            mock_conf.assert_not_called()
+            mock_creds.refresh.assert_called_once()
+            assert any(
+                "certificate has not changed" in str(call)
+                for call in mock_info.call_args_list
+            )
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_401_refresh_raises_invalid_operation_returns_401(self):
+        """Verifies that exceptions.InvalidOperation during refresh is caught and returns the 401 response."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(
+            side_effect=exceptions.InvalidOperation("Invalid operation")
+        )
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_401.close = mock.AsyncMock()
+
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp_401)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+
+        resp = await session.request(
+            "GET", "https://example.com"
+        )
+        assert resp == mock_resp_401
+        mock_creds.refresh.assert_called_once()
+        await session.close()

From 367585814e7fc6f1d9dfd9e74ca6386a4a57ef25 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 19:40:53 +0000
Subject: [PATCH 05/22] fix(auth): make configure_mtls_channel idempotent for
 repeated default calls and preserve metadata state on error

- Distinguish between idempotent default calls and explicit reconfigurations in configure_mtls_channel
- Preserve self._is_mtls and self._cached_cert on configuration failure to keep metadata in sync with the active _auth_request
- Add unit tests verifying configure_mtls_channel idempotency and state preservation
---
 .../google/auth/aio/transport/sessions.py     | 17 +++-
 .../tests/transport/aio/test_sessions_mtls.py | 87 ++++++++++++++++++-
 2 files changed, 97 insertions(+), 7 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index d85ee18d769f..86368bdfc34c 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -193,7 +193,20 @@ async def configure_mtls_channel(self, client_cert_callback=None):
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
         """
-        if self._mtls_init_task is None or self._mtls_init_task.done():
+        is_explicit_reconfig = (
+            client_cert_callback is not None
+            and client_cert_callback != self._client_cert_callback
+        )
+        task_failed = (
+            self._mtls_init_task is not None
+            and self._mtls_init_task.done()
+            and (
+                self._mtls_init_task.cancelled()
+                or self._mtls_init_task.exception() is not None
+            )
+        )
+
+        if self._mtls_init_task is None or is_explicit_reconfig or task_failed:
             self._client_cert_callback = client_cert_callback
 
             async def _do_configure():
@@ -255,8 +268,6 @@ async def _do_configure():
                         self._cached_cert = None
 
                 except Exception as caught_exc:
-                    self._is_mtls = False
-                    self._cached_cert = None
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc
 
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 7d03dbbee120..d8e12494b73c 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -183,7 +183,7 @@ async def test_configure_mtls_channel_custom_request(self):
             await session.close()
 
     @pytest.mark.asyncio
-    async def test_configure_mtls_channel_exception_resets_flag(self):
+    async def test_configure_mtls_channel_exception_preserves_flag(self):
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
         session = sessions.AsyncAuthorizedSession(mock_creds)
         session._is_mtls = True
@@ -199,9 +199,9 @@ async def test_configure_mtls_channel_exception_resets_flag(self):
             ),
             pytest.raises(exceptions.MutualTLSChannelError),
         ):
-            await session.configure_mtls_channel()
-        assert session.is_mtls is False
-        assert session._cached_cert == None
+            await session.configure_mtls_channel(lambda: (b"new_cert", b"new_key"))
+        assert session.is_mtls is True
+        assert session._cached_cert == b"old_cert"
         await session.close()
 
     @pytest.mark.asyncio
@@ -1471,3 +1471,82 @@ async def test_401_refresh_raises_invalid_operation_returns_401(self):
         assert resp == mock_resp_401
         mock_creds.refresh.assert_called_once()
         await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_idempotent_when_called_repeatedly(self):
+        """Tests that calling configure_mtls_channel repeatedly without a new callback reuses the existing task."""
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert_bytes", b"key_bytes"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            await session.configure_mtls_channel()
+            assert session.is_mtls is True
+            first_auth_req = session._auth_request
+            first_task = session._mtls_init_task
+
+            # Call configure_mtls_channel again without new callback
+            await session.configure_mtls_channel()
+            assert session._auth_request is first_auth_req
+            assert session._mtls_init_task is first_task
+
+            # Call configure_mtls_channel with a new callback - should reconfigure
+            new_callback = lambda: (b"new_cert", b"new_key")
+            await session.configure_mtls_channel(new_callback)
+            assert session._auth_request is not first_auth_req
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_exception_preserves_existing_mtls_state(self):
+        """Tests that an exception during re-configuration does not clear existing is_mtls and cached_cert."""
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert_v1", b"key_v1"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            await session.configure_mtls_channel()
+            assert session.is_mtls is True
+            assert session._cached_cert == b"cert_v1"
+            first_auth_req = session._auth_request
+
+            # Now attempt reconfiguring with a failing callback/context
+            with (
+                mock.patch(
+                    "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                    side_effect=RuntimeError("Reconfig failure"),
+                ),
+                pytest.raises(exceptions.MutualTLSChannelError),
+            ):
+                await session.configure_mtls_channel(lambda: (b"cert_v2", b"key_v2"))
+
+            # Session should still retain its previous mTLS state and active auth_request
+            assert session.is_mtls is True
+            assert session._cached_cert == b"cert_v1"
+            assert session._auth_request is first_auth_req
+            await session.close()

From 5aef5c3ca9b65c33135b71b90aa53190cb8ba98b Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 19:54:29 +0000
Subject: [PATCH 06/22] test(auth): make
 test_401_retry_raises_timeout_before_subsequent_retry robust across Python
 versions

---
 .../tests/transport/aio/test_sessions_mtls.py         | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index d8e12494b73c..9afbdbca7e1d 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1101,12 +1101,15 @@ async def fake_refresh(auth_request):
             mock_creds, auth_request=mock_auth_request
         )
 
-        time_calls = [0.0, 0.0, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 100.0]
+        after_refresh_count = 0
 
         def mock_time():
-            if time_calls:
-                return time_calls.pop(0)
-            return 100.0
+            nonlocal after_refresh_count
+            if mock_creds.refresh.called:
+                after_refresh_count += 1
+                if after_refresh_count >= 2:
+                    return 100.0
+            return 0.1
 
         with mock.patch("time.monotonic", side_effect=mock_time):
             with pytest.raises(

From 96ec6228c2b244b9878970f2739adc55e3d767da Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 20:06:03 +0000
Subject: [PATCH 07/22] fix(auth): allow mTLS retry when credentials raise
 NotImplementedError on refresh

- In AsyncAuthorizedSession.request()'s _recover_auth_state(), do not return 401 response on NotImplementedError so that mTLS reconfiguration can fall through to retry
- Add unit test test_cert_rotation_credential_refresh_not_implemented_retries
---
 .../google/auth/aio/transport/sessions.py     |  1 -
 .../tests/transport/aio/test_sessions_mtls.py | 52 +++++++++++++++++++
 2 files changed, 52 insertions(+), 1 deletion(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 86368bdfc34c..c914cdaefb6e 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -496,7 +496,6 @@ async def _recover_auth_state():
                                     _LOGGER.debug(
                                         "Credentials do not implement refresh()."
                                     )
-                                    return response
                                 except (
                                     exceptions.RefreshError,
                                     exceptions.InvalidOperation,
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 9afbdbca7e1d..911cf464ec95 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1553,3 +1553,55 @@ async def test_configure_mtls_channel_exception_preserves_existing_mtls_state(se
             assert session._cached_cert == b"cert_v1"
             assert session._auth_request is first_auth_req
             await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_credential_refresh_not_implemented_retries(self):
+        """Validate credentials that raise NotImplementedError on refresh()
+        still trigger a retry after mTLS reconfiguration, not return the 401."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(side_effect=NotImplementedError)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_401.close = mock.AsyncMock()
+
+        mock_resp_200 = mock.Mock()
+        mock_resp_200.status_code = http_client.OK
+        mock_resp_200.close = mock.AsyncMock()
+
+        mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200])
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        with mock.patch(
+            "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
+            new_callable=mock.AsyncMock,
+        ) as mock_check:
+            with mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf:
+                mock_check.return_value = (
+                    b"new_cert",
+                    b"new_key",
+                    b"old_fp",
+                    b"new_fp",
+                )
+
+                resp = await session.request(
+                    "GET", "https://pubsub.mtls.googleapis.com/test"
+                )
+
+                # Validate that the handler falls through to `return None`
+                # on NotImplementedError in order to signal retry.
+                assert resp == mock_resp_200
+                mock_conf.assert_called_once()
+                mock_creds.refresh.assert_called_once()
+                assert mock_auth_req.call_count == 2
+                mock_resp_401.close.assert_called_once()
+
+        await session.close()

From 688f76fe3b1a837e278ca8e97f577f12863b8ceb Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 20:25:11 +0000
Subject: [PATCH 08/22] style(auth): format with black and fix flake8 style
 issues

---
 .../google/auth/aio/transport/sessions.py        | 13 ++++++++-----
 .../tests/transport/aio/test_sessions_mtls.py    | 16 +++++++---------
 2 files changed, 15 insertions(+), 14 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index c914cdaefb6e..8333ae3b0f0c 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -450,10 +450,10 @@ async def _recover_auth_state():
                                                     )
                                                     await self.configure_mtls_channel(
                                                         lambda: (
-                                                             call_cert_bytes,
-                                                             call_key_bytes,
-                                                         )
-                                                     )
+                                                            call_cert_bytes,
+                                                            call_key_bytes,
+                                                        )
+                                                    )
                                                     channel_reconfigured = True
                                                 except Exception as e:
                                                     _LOGGER.error(
@@ -485,7 +485,10 @@ async def _recover_auth_state():
 
                         async with self._refresh_lock:
                             # Check if another task already refreshed credentials while we were waiting
-                            if not channel_reconfigured and self._refresh_counter > refresh_counter_at_error:
+                            if (
+                                not channel_reconfigured
+                                and self._refresh_counter > refresh_counter_at_error
+                            ):
                                 _LOGGER.debug(
                                     "Credentials were already refreshed by a concurrent task. Skipping duplicate refresh."
                                 )
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 911cf464ec95..77f6eca6e0e3 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -179,7 +179,7 @@ async def test_configure_mtls_channel_custom_request(self):
         ):
             await session.configure_mtls_channel()
             assert session.is_mtls is False
-            assert session._cached_cert == None
+            assert session._cached_cert is None
             await session.close()
 
     @pytest.mark.asyncio
@@ -226,7 +226,7 @@ async def test_configure_mtls_channel_transport_error_resets_flag(self):
         ):
             await session.configure_mtls_channel()
         assert session.is_mtls is False
-        assert session._cached_cert == None
+        assert session._cached_cert is None
         await session.close()
 
     @pytest.mark.asyncio
@@ -1425,9 +1425,7 @@ async def test_401_cert_check_without_cached_cert_skips_reconfiguration(self):
                 new_callable=mock.AsyncMock,
                 return_value=(True, b"cert_bytes", b"key_bytes"),
             ),
-            mock.patch(
-                "google.auth._agent_identity_utils.parse_certificate"
-            ),
+            mock.patch("google.auth._agent_identity_utils.parse_certificate"),
             mock.patch(
                 "google.auth._agent_identity_utils.calculate_certificate_fingerprint",
                 return_value="FINGERPRINT_1",
@@ -1468,9 +1466,7 @@ async def test_401_refresh_raises_invalid_operation_returns_401(self):
             mock_creds, auth_request=mock_auth_req
         )
 
-        resp = await session.request(
-            "GET", "https://example.com"
-        )
+        resp = await session.request("GET", "https://example.com")
         assert resp == mock_resp_401
         mock_creds.refresh.assert_called_once()
         await session.close()
@@ -1507,7 +1503,9 @@ async def test_configure_mtls_channel_idempotent_when_called_repeatedly(self):
             assert session._mtls_init_task is first_task
 
             # Call configure_mtls_channel with a new callback - should reconfigure
-            new_callback = lambda: (b"new_cert", b"new_key")
+            def new_callback():
+                return b"new_cert", b"new_key"
+
             await session.configure_mtls_channel(new_callback)
             assert session._auth_request is not first_auth_req
             await session.close()

From 54eb0bc0c0d39e4c3cb22f24b52936d1d93c16e0 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 20:30:57 +0000
Subject: [PATCH 09/22] test(auth): add unit test for consecutive mTLS
 certificate rotations

---
 .../tests/transport/aio/test_sessions_mtls.py | 58 +++++++++++++++++++
 1 file changed, 58 insertions(+)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 77f6eca6e0e3..bff8e6c1420f 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1603,3 +1603,61 @@ async def test_cert_rotation_credential_refresh_not_implemented_retries(self):
                 mock_resp_401.close.assert_called_once()
 
         await session.close()
+
+    @pytest.mark.asyncio
+    async def test_401_mtls_consecutive_multi_rotation(self):
+        """Verifies that multiple consecutive rotations (v1 -> v2 -> v3) succeed."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        session = sessions.AsyncAuthorizedSession(mock_creds)
+        session._is_mtls = True
+        session._cached_cert = b"cert_v1"
+
+        # Rotation 1: v1 -> v2
+        with mock.patch(
+            "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
+            new_callable=mock.AsyncMock,
+            return_value=(b"cert_v2", b"key_v2", b"fp1", b"fp2"),
+        ):
+            with mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf:
+                mock_auth = mock.AsyncMock(
+                    side_effect=[
+                        mock.Mock(status_code=401, close=mock.AsyncMock()),
+                        mock.Mock(status_code=200, close=mock.AsyncMock()),
+                    ]
+                )
+                session._auth_request = mock_auth
+                await session.request("GET", "https://pubsub.mtls.googleapis.com/test")
+                mock_conf.assert_called_once()
+                assert session._client_cert_callback is None
+
+        session._cached_cert = b"cert_v2"
+
+        # Rotation 2: v2 -> v3 (Must still have client_cert_callback == None to read disk)
+        with mock.patch(
+            "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
+            new_callable=mock.AsyncMock,
+            return_value=(b"cert_v3", b"key_v3", b"fp2", b"fp3"),
+        ) as mock_check:
+            with mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf:
+                mock_auth = mock.AsyncMock(
+                    side_effect=[
+                        mock.Mock(status_code=401, close=mock.AsyncMock()),
+                        mock.Mock(status_code=200, close=mock.AsyncMock()),
+                    ]
+                )
+                session._auth_request = mock_auth
+                await session.request("GET", "https://pubsub.mtls.googleapis.com/test")
+                # Verify check was called with callback=None (allowing disk read)
+                mock_check.assert_called_with(b"cert_v2", None)
+                mock_conf.assert_called_once()
+                assert session._client_cert_callback is None
+
+        await session.close()
+

From 118785bd9a84623bb0078b96a4f48b45a057d396 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 20:38:06 +0000
Subject: [PATCH 10/22] test(auth): pass mock_auth_request in cancellation test
 to avoid unmocked transport

---
 .../google-auth/tests/transport/aio/test_sessions_mtls.py    | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index bff8e6c1420f..5b5d097deb17 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1245,7 +1245,10 @@ async def test_request_cancellation_propagates_and_leaves_mtls_init_running(self
         in the background.
         """
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
-        session = sessions.AsyncAuthorizedSession(mock_creds)
+        mock_auth_request = mock.AsyncMock(spec=transport.Request)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_request
+        )
         init_started = asyncio.Event()
         init_can_finish = asyncio.Event()
 

From e0467602e9030674b7064365cd2acbdd3e9d73ef Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 10 Sep 2026 20:56:32 +0000
Subject: [PATCH 11/22] test(auth): allow timeout_guard error message in
 test_401_retry_raises_timeout_before_subsequent_retry

---
 packages/google-auth/tests/transport/aio/test_sessions_mtls.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 5b5d097deb17..a89a261ae9b3 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1114,7 +1114,7 @@ def mock_time():
         with mock.patch("time.monotonic", side_effect=mock_time):
             with pytest.raises(
                 exceptions.TimeoutError,
-                match="Timeout exceeded before retrying the request",
+                match=r"(Timeout exceeded before retrying the request|Context manager exceeded the configured timeout)",
             ):
                 await session.request(
                     "GET", "https://example.com", max_allowed_time=1.0
@@ -1663,4 +1663,3 @@ async def test_401_mtls_consecutive_multi_rotation(self):
                 assert session._client_cert_callback is None
 
         await session.close()
-

From aaa4a310cd76cda6958b820100f9fa7124ca8f91 Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 04:40:17 +0000
Subject: [PATCH 12/22] fix: Patch code logic and update tests

---
 .../google/auth/aio/transport/sessions.py     |  39 +++--
 .../tests/transport/aio/test_sessions_mtls.py | 140 +++++++++++++++++-
 2 files changed, 163 insertions(+), 16 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 8333ae3b0f0c..ec1ff26421d9 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -168,7 +168,9 @@ def __init__(
         self._refresh_lock: Optional[asyncio.Lock] = None
         self._refresh_counter = 0
 
-    async def configure_mtls_channel(self, client_cert_callback=None):
+    async def configure_mtls_channel(
+        self, client_cert_callback=None, force: bool = False
+    ):
         """Configure the client certificate and key for SSL connection.
 
         This method configures mTLS if client certificates are explicitly enabled
@@ -188,15 +190,15 @@ async def configure_mtls_channel(self, client_cert_callback=None):
                 key bytes both in PEM format.
                 If the callback is None, application default SSL credentials
                 will be used.
+            force (bool):
+                Whether to force reconfiguration even if the channel is already configured
+                with the same callback.
 
         Raises:
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
         """
-        is_explicit_reconfig = (
-            client_cert_callback is not None
-            and client_cert_callback != self._client_cert_callback
-        )
+        is_explicit_reconfig = client_cert_callback != self._client_cert_callback
         task_failed = (
             self._mtls_init_task is not None
             and self._mtls_init_task.done()
@@ -206,7 +208,12 @@ async def configure_mtls_channel(self, client_cert_callback=None):
             )
         )
 
-        if self._mtls_init_task is None or is_explicit_reconfig or task_failed:
+        if self._mtls_init_task is None or is_explicit_reconfig or task_failed or force:
+            if self._mtls_init_task is not None and not self._mtls_init_task.done():
+                try:
+                    await self._mtls_init_task
+                except Exception:
+                    pass
             self._client_cert_callback = client_cert_callback
 
             async def _do_configure():
@@ -409,7 +416,6 @@ async def _recover_auth_state():
                                     ):
                                         pass
                                     else:
-                                        check_passed = False
                                         try:
                                             (
                                                 call_cert_bytes,
@@ -420,7 +426,6 @@ async def _recover_auth_state():
                                                 self._cached_cert,
                                                 self._client_cert_callback,
                                             )
-                                            check_passed = True
                                         except (
                                             exceptions.ClientCertError,
                                             exceptions.MutualTLSChannelError,
@@ -478,8 +483,8 @@ async def _recover_auth_state():
                                                         "Skipping reconfiguration of mTLS channel because the client"
                                                         " certificate has not changed."
                                                     )
-                                        if check_passed:
-                                            self._mtls_check_counter += 1
+                                        # Always increment so waiting tasks skip the check block
+                                        self._mtls_check_counter += 1
                         if self._refresh_lock is None:
                             self._refresh_lock = asyncio.Lock()
 
@@ -499,10 +504,16 @@ async def _recover_auth_state():
                                     _LOGGER.debug(
                                         "Credentials do not implement refresh()."
                                     )
-                                except (
-                                    exceptions.RefreshError,
-                                    exceptions.InvalidOperation,
-                                ) as e:
+                                    if not channel_reconfigured:
+                                        return response
+                                except exceptions.InvalidOperation as e:
+                                    _LOGGER.debug(
+                                        "Credentials cannot be refreshed: %s",
+                                        e,
+                                    )
+                                    if not channel_reconfigured:
+                                        return response
+                                except exceptions.RefreshError as e:
                                     _LOGGER.debug(
                                         "Credential refresh failed, returning 401 response. Error: %s",
                                         e,
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index a89a261ae9b3..9264f15dddab 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1181,8 +1181,8 @@ async def slow_failing_check(*args, **kwargs):
             )
 
             assert results == [mock_resp_200_1, mock_resp_200_2]
-            assert mock_check.call_count == 2
-            assert session._mtls_check_counter == 0
+            assert mock_check.call_count == 1
+            assert session._mtls_check_counter == 1
             assert mock_conf.call_count == 0
             assert mock_creds.refresh.call_count == 1
 
@@ -1663,3 +1663,139 @@ async def test_401_mtls_consecutive_multi_rotation(self):
                 assert session._client_cert_callback is None
 
         await session.close()
+
+    @pytest.mark.asyncio
+    async def test_non_mtls_not_implemented_refresh_returns_401_without_retry(self):
+        """Verifies that non-mTLS 401s on credentials that raise NotImplementedError
+        return the 401 immediately without retrying."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(side_effect=NotImplementedError)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_401.close = mock.AsyncMock()
+
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp_401)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+
+        resp = await session.request("GET", "https://example.com")
+        assert resp == mock_resp_401
+        assert mock_auth_req.call_count == 1
+        mock_creds.refresh.assert_called_once()
+        mock_resp_401.close.assert_not_called()
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_credential_refresh_invalid_operation_retries(self):
+        """Verifies that when mTLS is reconfigured, credentials raising InvalidOperation
+        (e.g., StaticCredentials) still retry on the reconfigured mTLS channel."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(
+            side_effect=exceptions.InvalidOperation("Static credentials")
+        )
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_401.close = mock.AsyncMock()
+
+        mock_resp_200 = mock.Mock()
+        mock_resp_200.status_code = http_client.OK
+        mock_resp_200.close = mock.AsyncMock()
+
+        mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200])
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        with (
+            mock.patch(
+                "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
+                new_callable=mock.AsyncMock,
+                return_value=(b"new_cert", b"new_key", b"old_fp", b"new_fp"),
+            ),
+            mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf,
+        ):
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+            assert resp == mock_resp_200
+            mock_conf.assert_called_once()
+            mock_creds.refresh.assert_called_once()
+            assert mock_auth_req.call_count == 2
+            mock_resp_401.close.assert_called_once()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_reverts_to_default_when_callback_none(self):
+        """Tests that passing callback=None when a callback was previously set reconfigures back to ADC."""
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            def custom_cb():
+                return b"c", b"k"
+
+            await session.configure_mtls_channel(custom_cb)
+            assert session._client_cert_callback is custom_cb
+            task1 = session._mtls_init_task
+
+            # Revert to default
+            await session.configure_mtls_channel(None)
+            assert session._client_cert_callback is None
+            assert session._mtls_init_task is not task1
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_force_reconfigures_same_callback(self):
+        """Tests that force=True reconfigures even if callback is identical."""
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            await session.configure_mtls_channel()
+            task1 = session._mtls_init_task
+
+            # Same callback with force=True
+            await session.configure_mtls_channel(force=True)
+            assert session._mtls_init_task is not task1
+            await session.close()

From 0be988d56970cb33792827d5f6ca9db03f489b0e Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 05:22:32 +0000
Subject: [PATCH 13/22] Update
 packages/google-auth/google/auth/aio/transport/sessions.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/google-auth/google/auth/aio/transport/sessions.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index ec1ff26421d9..b08a7daaf951 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -336,7 +336,7 @@ async def request(
                 # allowing the request to fail naturally elsewhere.
                 pass
             except asyncio.CancelledError:
-                if self._mtls_init_task.cancelled():
+                if self._mtls_init_task and self._mtls_init_task.cancelled():
                     pass
                 else:
                     raise

From 2a9838cee4a09dd8f90f9404f4b6993c8432c134 Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 05:23:29 +0000
Subject: [PATCH 14/22] Update
 packages/google-auth/google/auth/aio/transport/sessions.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/google-auth/google/auth/aio/transport/sessions.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index b08a7daaf951..7bce77cd9571 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -210,9 +210,10 @@ async def configure_mtls_channel(
 
         if self._mtls_init_task is None or is_explicit_reconfig or task_failed or force:
             if self._mtls_init_task is not None and not self._mtls_init_task.done():
+                self._mtls_init_task.cancel()
                 try:
                     await self._mtls_init_task
-                except Exception:
+                except (Exception, asyncio.CancelledError):
                     pass
             self._client_cert_callback = client_cert_callback
 

From ddf2c31f9d0f231b1334269b45db298b979cfe1d Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 05:23:46 +0000
Subject: [PATCH 15/22] Update
 packages/google-auth/tests/transport/aio/test_sessions_mtls.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/google-auth/tests/transport/aio/test_sessions_mtls.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 9264f15dddab..4918bdaed090 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1111,7 +1111,7 @@ def mock_time():
                     return 100.0
             return 0.1
 
-        with mock.patch("time.monotonic", side_effect=mock_time):
+        with mock.patch("google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time):
             with pytest.raises(
                 exceptions.TimeoutError,
                 match=r"(Timeout exceeded before retrying the request|Context manager exceeded the configured timeout)",

From 449d6812a72ed54b452a5c26c805d1b77150db0a Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 05:24:01 +0000
Subject: [PATCH 16/22] Update
 packages/google-auth/tests/transport/aio/test_sessions_mtls.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/google-auth/tests/transport/aio/test_sessions_mtls.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 4918bdaed090..06513b3083f3 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1069,7 +1069,7 @@ def mock_time():
                 return 100.0
             return 0.1
 
-        with mock.patch("time.monotonic", side_effect=mock_time):
+        with mock.patch("google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time):
             with pytest.raises(
                 exceptions.TimeoutError,
                 match="Timeout exceeded before credential refresh could begin",

From 87d28eb76930db4bb06fa0f23f42af40371861ec Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 05:33:31 +0000
Subject: [PATCH 17/22] fix(auth): add type annotation and assertion for
 _mtls_init_task to resolve mypy error

---
 packages/google-auth/google/auth/aio/transport/sessions.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 7bce77cd9571..8b129e30028f 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -154,7 +154,7 @@ def __init__(
         if not _auth_request and AIOHTTP_INSTALLED:
             _auth_request = AiohttpRequest()
         self._is_mtls = False
-        self._mtls_init_task = None
+        self._mtls_init_task: Optional[asyncio.Task] = None
         self._cached_cert = None
         self._client_cert_callback = None
         self._old_auth_requests: list[transport.Request] = []
@@ -281,6 +281,7 @@ async def _do_configure():
 
             self._mtls_init_task = asyncio.create_task(_do_configure())
 
+        assert self._mtls_init_task is not None
         return await asyncio.shield(self._mtls_init_task)
 
     async def request(

From ff80fa6514f82924e81df87984a4d77778d4156a Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sat, 12 Sep 2026 05:39:34 +0000
Subject: [PATCH 18/22] fix lint

---
 .../google-auth/tests/transport/aio/test_sessions_mtls.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 06513b3083f3..c82b4e6a91db 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -1069,7 +1069,9 @@ def mock_time():
                 return 100.0
             return 0.1
 
-        with mock.patch("google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time):
+        with mock.patch(
+            "google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time
+        ):
             with pytest.raises(
                 exceptions.TimeoutError,
                 match="Timeout exceeded before credential refresh could begin",
@@ -1111,7 +1113,9 @@ def mock_time():
                     return 100.0
             return 0.1
 
-        with mock.patch("google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time):
+        with mock.patch(
+            "google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time
+        ):
             with pytest.raises(
                 exceptions.TimeoutError,
                 match=r"(Timeout exceeded before retrying the request|Context manager exceeded the configured timeout)",

From a6aa06b07a021e4435653fda10c231d9234035c1 Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sun, 13 Sep 2026 05:20:03 +0000
Subject: [PATCH 19/22] fix(auth): address review findings in async mTLS
 session concurrency

Addresses the review on #18355 plus two self-identified gaps.

Reviewer findings:

- Rotation state was tracked in a coroutine-local `channel_reconfigured`
  flag. A concurrent request that skipped the rotation check (because
  another request had already bumped `_mtls_check_counter`) still saw
  `False`, so with non-refreshable credentials it returned the stale 401
  instead of retrying on the freshly rotated channel. Replaced with a
  session-level `_mtls_reconfig_counter` snapshotted per request.

- `except (Exception, asyncio.CancelledError): pass` in
  `configure_mtls_channel()` swallowed the coroutine's own cancellation,
  and two callers could both spawn `_do_configure()`. Task creation is now
  serialized under `_mtls_init_lock`, and the retired task is awaited via
  `asyncio.wait()`, which does not absorb caller cancellation.

- The rotation path temporarily swapped `self._client_cert_callback` with
  a lambda, which is shared state that could be observed or clobbered by
  concurrent callers and leaked on timeout. Replaced with an internal
  `_cert_key_override` argument; the user callback is never mutated.

- Leaving mTLS reset `_is_mtls`/`_cached_cert` but kept the old mTLS
  transport, so later requests still presented the retired client
  certificate. `_reset_non_mtls_state()` now retires it and installs a
  fresh transport. Dropped the dead `is_mtls = False` assignment.

- `await asyncio.shield(self._mtls_init_task)` raised a spurious
  `CancelledError` when a concurrent reconfiguration replaced the task,
  and a background failure nobody awaited surfaced as "Task exception was
  never retrieved". Callers now follow the replacement task (re-read under
  `_mtls_init_lock`, which the reconfiguring coroutine holds across cancel
  and create) and a done-callback retrieves the exception.

- `request()` re-read `self._mtls_init_task` after awaiting, inspecting a
  possibly-replaced task. Replaced with the loop suggested in review.

- `test_401_mtls_consecutive_multi_rotation` mocked
  `configure_mtls_channel` outright, so it could not detect callback
  clobbering. It now patches only the low-level helpers so the real
  reconfiguration path runs, and uses a non-None user callback so an
  overwrite is actually observable (verified by mutation testing).

Self-identified gaps:

- `assert self._mtls_init_task is not None` was load-bearing for mypy but
  is stripped under `python -O`. Removed in favour of a typed local and an
  explicit error.

- Deriving `channel_reconfigured` from a session counter initially made
  every concurrent request force its own refresh, defeating refresh
  de-duplication. Refreshes are now de-duplicated against the later of the
  request's own 401 and the most recent reconfiguration, so exactly one
  refresh runs against a rotated channel.

Tests: 326 mTLS-related tests pass; full suite 2148 passed with 9
pre-existing failures unrelated to this change (verified against a clean
baseline). black, flake8 and mypy are clean.

TAG=agy
CONV=547fc656-aa8a-4bda-83c8-d57057515ec3
---
 .../google/auth/aio/transport/sessions.py     | 346 +++++++++++------
 .../tests/transport/aio/test_sessions_mtls.py | 350 +++++++++++++++---
 2 files changed, 532 insertions(+), 164 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 8b129e30028f..54d6394e0b9c 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -20,7 +20,7 @@
 import inspect
 import logging
 import time
-from typing import Mapping, Optional, TYPE_CHECKING, Union
+from typing import Mapping, Optional, Tuple, TYPE_CHECKING, Union
 import urllib.parse
 import warnings
 
@@ -58,6 +58,19 @@
     AIOHTTP_INSTALLED = False
 
 
+def _retrieve_task_exception(task: "asyncio.Task") -> None:
+    """Mark a finished task's exception as retrieved.
+
+    Used as a done-callback so that a background mTLS initialization failure
+    that nobody ends up awaiting (for example, because the only caller timed
+    out) does not later surface as an unraisable
+    "Task exception was never retrieved" warning during garbage collection.
+    Callers that *do* await the task still observe the exception normally.
+    """
+    if not task.cancelled():
+        task.exception()
+
+
 @asynccontextmanager
 async def timeout_guard(timeout):
     """
@@ -165,11 +178,61 @@ def __init__(
         self._auth_request = _auth_request
         self._mtls_rotation_lock: Optional[asyncio.Lock] = None
         self._mtls_check_counter = 0
+        # Incremented every time the mTLS channel is successfully reconfigured.
+        # Unlike a coroutine-local flag, this lets a request that skipped the
+        # rotation check (because a concurrent request already performed it)
+        # still observe that the channel changed since its own 401.
+        self._mtls_reconfig_counter = 0
+        # Value of `_refresh_counter` at the moment of the most recent
+        # reconfiguration. A credential refresh is only considered redundant
+        # if it happened *after* the latest channel change, since a
+        # certificate-bound token minted on the old channel is not valid on
+        # the new one.
+        self._refresh_counter_at_last_reconfig = 0
+        # Serializes the decision to create a new mTLS initialization task so
+        # that two concurrent callers cannot both spawn `_do_configure()`.
+        self._mtls_init_lock: Optional[asyncio.Lock] = None
         self._refresh_lock: Optional[asyncio.Lock] = None
         self._refresh_counter = 0
 
+    async def _trim_old_auth_requests(self) -> None:
+        """Close retired transports, keeping at most the two most recent."""
+        while len(self._old_auth_requests) > 2:
+            oldest_auth_request = self._old_auth_requests.pop(0)
+            try:
+                if hasattr(oldest_auth_request, "close"):
+                    res = oldest_auth_request.close()
+                    if inspect.isawaitable(res):
+                        await res
+            except Exception:
+                pass
+
+    async def _reset_non_mtls_state(self) -> None:
+        """Clear mTLS state, retiring a library-owned mTLS transport.
+
+        If the session previously installed its own mTLS-enabled transport,
+        that transport still presents the old client certificate. Simply
+        clearing the flags would leave subsequent requests sending a stale
+        cert, so the transport is retired and replaced with a fresh non-mTLS
+        one. A caller-supplied custom transport is never replaced.
+        """
+        was_mtls = self._is_mtls
+        self._is_mtls = False
+        self._cached_cert = None
+        if (
+            was_mtls
+            and AIOHTTP_INSTALLED
+            and isinstance(self._auth_request, AiohttpRequest)
+        ):
+            self._old_auth_requests.append(self._auth_request)
+            self._auth_request = AiohttpRequest()
+            await self._trim_old_auth_requests()
+
     async def configure_mtls_channel(
-        self, client_cert_callback=None, force: bool = False
+        self,
+        client_cert_callback=None,
+        force: bool = False,
+        _cert_key_override: Optional[Tuple[bytes, bytes]] = None,
     ):
         """Configure the client certificate and key for SSL connection.
 
@@ -193,96 +256,151 @@ async def configure_mtls_channel(
             force (bool):
                 Whether to force reconfiguration even if the channel is already configured
                 with the same callback.
+            _cert_key_override (Optional[Tuple[bytes, bytes]]):
+                Internal use only. An explicit (cert, key) pair to install,
+                bypassing ``client_cert_callback`` resolution. Used by the
+                certificate-rotation path so it does not have to temporarily
+                mutate ``self._client_cert_callback``, which would otherwise
+                be visible to (and clobber) concurrent callers. When set, the
+                channel is always reconfigured and the user-supplied callback
+                is left untouched.
 
         Raises:
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
         """
-        is_explicit_reconfig = client_cert_callback != self._client_cert_callback
-        task_failed = (
-            self._mtls_init_task is not None
-            and self._mtls_init_task.done()
-            and (
-                self._mtls_init_task.cancelled()
-                or self._mtls_init_task.exception() is not None
-            )
-        )
-
-        if self._mtls_init_task is None or is_explicit_reconfig or task_failed or force:
-            if self._mtls_init_task is not None and not self._mtls_init_task.done():
-                self._mtls_init_task.cancel()
-                try:
-                    await self._mtls_init_task
-                except (Exception, asyncio.CancelledError):
-                    pass
-            self._client_cert_callback = client_cert_callback
-
-            async def _do_configure():
-                # Run the blocking check in an executor
-                use_client_cert = await mtls._run_in_executor(
-                    google.auth.transport._mtls_helper.check_use_client_cert
+        if self._mtls_init_lock is None:
+            self._mtls_init_lock = asyncio.Lock()
+        init_lock = self._mtls_init_lock
+
+        # Serialize the decide-and-create step so two concurrent callers cannot
+        # both spawn `_do_configure()`. The lock is released before awaiting the
+        # task itself, so a slow configuration does not block unrelated callers.
+        async with init_lock:
+            if _cert_key_override is not None:
+                needs_reconfig = True
+            else:
+                is_explicit_reconfig = (
+                    client_cert_callback != self._client_cert_callback
+                )
+                task_failed = (
+                    self._mtls_init_task is not None
+                    and self._mtls_init_task.done()
+                    and (
+                        self._mtls_init_task.cancelled()
+                        or self._mtls_init_task.exception() is not None
+                    )
+                )
+                needs_reconfig = (
+                    self._mtls_init_task is None
+                    or is_explicit_reconfig
+                    or task_failed
+                    or force
                 )
-                if not use_client_cert:
-                    return
-
-                try:
-                    (
-                        is_mtls,
-                        cert,
-                        key,
-                    ) = await mtls.get_client_cert_and_key(client_cert_callback)
-
-                    if is_mtls:
-                        # Re-create the auth request with the new SSL context
-                        if AIOHTTP_INSTALLED and isinstance(
-                            self._auth_request, AiohttpRequest
-                        ):
-                            ssl_context = await mtls._run_in_executor(
-                                mtls.make_client_cert_ssl_context, cert, key
-                            )
-                            connector = aiohttp.TCPConnector(ssl=ssl_context)
-                            new_session = aiohttp.ClientSession(connector=connector)
-
-                            old_auth_request = self._auth_request
-                            self._auth_request = AiohttpRequest(session=new_session)
-                            self._is_mtls = True
-                            self._cached_cert = cert
-                            self._old_auth_requests.append(old_auth_request)
 
-                            while len(self._old_auth_requests) > 2:
-                                oldest_auth_request = self._old_auth_requests.pop(0)
-                                try:
-                                    if hasattr(oldest_auth_request, "close"):
-                                        res = oldest_auth_request.close()
-                                        if inspect.isawaitable(res):
-                                            await res
-                                except Exception:
-                                    pass
+            if not needs_reconfig:
+                task = self._mtls_init_task
+            else:
+                old_task = self._mtls_init_task
+                if old_task is not None and not old_task.done():
+                    old_task.cancel()
+                    # `asyncio.wait` does not re-raise the awaited task's
+                    # exception, and does not convert that task's cancellation
+                    # into ours -- while still letting a cancellation targeted
+                    # at *this* coroutine propagate normally.
+                    await asyncio.wait({old_task})
+                if _cert_key_override is None:
+                    # Only a user-driven call may change the stored callback.
+                    # The internal rotation path leaves it untouched.
+                    self._client_cert_callback = client_cert_callback
+
+                async def _do_configure():
+                    # Run the blocking check in an executor
+                    use_client_cert = await mtls._run_in_executor(
+                        google.auth.transport._mtls_helper.check_use_client_cert
+                    )
+                    if not use_client_cert:
+                        return
 
+                    try:
+                        if _cert_key_override is not None:
+                            is_mtls = True
+                            cert, key = _cert_key_override
                         else:
-                            is_mtls = False
-                            self._is_mtls = False
-                            self._cached_cert = None
-                            warnings.warn(
-                                "Attempted to establish mTLS, but a custom async transport was provided. "
-                                "google-auth cannot automatically configure custom transports for mTLS. "
-                                "Falling back to standard TLS. If your custom transport is not manually "
-                                "configured for mTLS, you may encounter 401 Unauthorized errors when "
-                                "using Certificate-Bound Tokens.",
-                                UserWarning,
-                            )
-                    else:
-                        self._is_mtls = False
-                        self._cached_cert = None
+                            (
+                                is_mtls,
+                                cert,
+                                key,
+                            ) = await mtls.get_client_cert_and_key(client_cert_callback)
+
+                        if is_mtls:
+                            # Re-create the auth request with the new SSL context
+                            if AIOHTTP_INSTALLED and isinstance(
+                                self._auth_request, AiohttpRequest
+                            ):
+                                ssl_context = await mtls._run_in_executor(
+                                    mtls.make_client_cert_ssl_context, cert, key
+                                )
+                                connector = aiohttp.TCPConnector(ssl=ssl_context)
+                                new_session = aiohttp.ClientSession(connector=connector)
 
-                except Exception as caught_exc:
-                    new_exc = exceptions.MutualTLSChannelError(caught_exc)
-                    raise new_exc from caught_exc
+                                old_auth_request = self._auth_request
+                                self._auth_request = AiohttpRequest(session=new_session)
+                                self._is_mtls = True
+                                self._cached_cert = cert
+                                self._old_auth_requests.append(old_auth_request)
+                                await self._trim_old_auth_requests()
 
-            self._mtls_init_task = asyncio.create_task(_do_configure())
+                            else:
+                                await self._reset_non_mtls_state()
+                                warnings.warn(
+                                    "Attempted to establish mTLS, but a custom async transport was provided. "
+                                    "google-auth cannot automatically configure custom transports for mTLS. "
+                                    "Falling back to standard TLS. If your custom transport is not manually "
+                                    "configured for mTLS, you may encounter 401 Unauthorized errors when "
+                                    "using Certificate-Bound Tokens.",
+                                    UserWarning,
+                                )
+                        else:
+                            await self._reset_non_mtls_state()
+
+                    except Exception as caught_exc:
+                        new_exc = exceptions.MutualTLSChannelError(caught_exc)
+                        raise new_exc from caught_exc
+
+                task = asyncio.create_task(_do_configure())
+                # If every awaiter goes away (e.g. the only caller timed out)
+                # a failure would otherwise surface as an unraisable
+                # "Task exception was never retrieved" warning at GC time.
+                task.add_done_callback(_retrieve_task_exception)
+                self._mtls_init_task = task
+
+            if task is None:  # pragma: no cover - defensive
+                raise exceptions.MutualTLSChannelError(
+                    "mTLS initialization task was not created."
+                )
 
-        assert self._mtls_init_task is not None
-        return await asyncio.shield(self._mtls_init_task)
+        while True:
+            try:
+                return await asyncio.shield(task)
+            except asyncio.CancelledError:
+                if not task.cancelled():
+                    # The cancellation targeted this caller rather than the
+                    # initialization task, so it must propagate.
+                    raise
+                # The task we were waiting on was cancelled by a concurrent
+                # reconfiguration. That coroutine holds `init_lock` across
+                # both the cancel and the creation of the replacement, so
+                # acquiring it here waits until the replacement is installed.
+                async with init_lock:
+                    current = self._mtls_init_task
+                if current is None or current is task:
+                    # Nobody installed a replacement (e.g. the session is
+                    # closing); surface the cancellation.
+                    raise
+                # Follow the replacement rather than reporting a cancellation
+                # this caller never requested.
+                task = current
 
     async def request(
         self,
@@ -330,18 +448,14 @@ async def request(
                 channel reconfiguration fails for any reason during certificate rotation.
         """
         _auth_retry_count = kwargs.pop("_auth_retry_count", 0)
-        if self._mtls_init_task and not self._mtls_init_task.done():
-            try:
-                await asyncio.shield(self._mtls_init_task)
-            except Exception:
-                # Suppress all exceptions from the background mTLS initialization task,
-                # allowing the request to fail naturally elsewhere.
-                pass
-            except asyncio.CancelledError:
-                if self._mtls_init_task and self._mtls_init_task.cancelled():
-                    pass
-                else:
-                    raise
+        # Wait for any in-flight mTLS initialization to settle. `asyncio.wait`
+        # neither re-raises the task's exception (the request should fail
+        # naturally elsewhere instead) nor turns that task's cancellation into
+        # ours, while a cancellation aimed at *this* coroutine still
+        # propagates. Looping re-reads the attribute in case a concurrent
+        # `configure_mtls_channel()` swapped in a replacement task.
+        while self._mtls_init_task is not None and not self._mtls_init_task.done():
+            await asyncio.wait({self._mtls_init_task})
         retries = _exponential_backoff.AsyncExponentialBackoff(
             total_attempts=total_attempts,
         )
@@ -349,6 +463,7 @@ async def request(
         start_time = time.monotonic()
         refresh_counter_at_error = self._refresh_counter
         check_counter_at_error = self._mtls_check_counter
+        reconfig_counter_at_error = self._mtls_reconfig_counter
         async with timeout_guard(max_allowed_time) as with_timeout:
             await with_timeout(
                 # Note: before_request will attempt to refresh credentials if expired.
@@ -394,7 +509,6 @@ async def request(
                     )
 
                     async def _recover_auth_state():
-                        channel_reconfigured = False
                         is_mtls_endpoint = False
                         if self._is_mtls:
                             hostname = urllib.parse.urlsplit(url).hostname
@@ -447,21 +561,29 @@ async def _recover_auth_state():
                                                 and cached_fingerprint
                                                 != current_cert_fingerprint
                                             ):
-                                                saved_callback = (
-                                                    self._client_cert_callback
-                                                )
                                                 try:
                                                     _LOGGER.info(
                                                         "Client certificate has changed, reconfiguring mTLS "
                                                         "channel."
                                                     )
+                                                    # Pass the rotated cert/key
+                                                    # directly rather than
+                                                    # temporarily swapping
+                                                    # `self._client_cert_callback`,
+                                                    # which is shared state and
+                                                    # could be observed (or
+                                                    # clobbered) by concurrent
+                                                    # callers.
                                                     await self.configure_mtls_channel(
-                                                        lambda: (
+                                                        _cert_key_override=(
                                                             call_cert_bytes,
                                                             call_key_bytes,
                                                         )
                                                     )
-                                                    channel_reconfigured = True
+                                                    self._refresh_counter_at_last_reconfig = (
+                                                        self._refresh_counter
+                                                    )
+                                                    self._mtls_reconfig_counter += 1
                                                 except Exception as e:
                                                     _LOGGER.error(
                                                         "Failed to reconfigure mTLS channel: %s",
@@ -470,10 +592,6 @@ async def _recover_auth_state():
                                                     raise exceptions.MutualTLSChannelError(
                                                         "Failed to reconfigure mTLS channel"
                                                     ) from e
-                                                finally:
-                                                    self._client_cert_callback = (
-                                                        saved_callback
-                                                    )
                                             else:
                                                 if current_cert_fingerprint is None:
                                                     _LOGGER.info(
@@ -487,15 +605,29 @@ async def _recover_auth_state():
                                                     )
                                         # Always increment so waiting tasks skip the check block
                                         self._mtls_check_counter += 1
+                        # Derived from session state rather than a local flag:
+                        # a concurrent request may have performed the rotation
+                        # on our behalf (we then skipped the check block), and
+                        # that request still needs to retry on the new channel.
+                        channel_reconfigured = (
+                            self._mtls_reconfig_counter > reconfig_counter_at_error
+                        )
+
                         if self._refresh_lock is None:
                             self._refresh_lock = asyncio.Lock()
 
                         async with self._refresh_lock:
-                            # Check if another task already refreshed credentials while we were waiting
-                            if (
-                                not channel_reconfigured
-                                and self._refresh_counter > refresh_counter_at_error
-                            ):
+                            # A refresh is redundant only if it happened after
+                            # both this request's 401 and the most recent
+                            # channel reconfiguration. Using the later of the
+                            # two keeps concurrent requests de-duplicated while
+                            # still guaranteeing at least one refresh against a
+                            # freshly rotated channel.
+                            refresh_baseline = max(
+                                refresh_counter_at_error,
+                                self._refresh_counter_at_last_reconfig,
+                            )
+                            if self._refresh_counter > refresh_baseline:
                                 _LOGGER.debug(
                                     "Credentials were already refreshed by a concurrent task. Skipping duplicate refresh."
                                 )
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index c82b4e6a91db..aebbfae40ec8 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -462,12 +462,14 @@ async def test_cert_rotation_success_and_retry(self):
 
             assert resp == mock_resp_200
             mock_conf.assert_called_once()
-            cb = (
-                mock_conf.call_args.args[0]
-                if mock_conf.call_args.args
-                else mock_conf.call_args.kwargs["client_cert_callback"]
+            # The rotation path passes the rotated cert/key explicitly rather
+            # than temporarily swapping the shared `_client_cert_callback`,
+            # which must therefore be left untouched.
+            assert mock_conf.call_args.kwargs["_cert_key_override"] == (
+                new_cert,
+                new_key,
             )
-            assert cb() == (new_cert, new_key)
+            assert session._client_cert_callback is None
             mock_creds.refresh.assert_called_once()
             assert mock_auth_req.call_count == 2
             mock_resp_401.close.assert_called_once()
@@ -642,12 +644,14 @@ async def test_psc_endpoint_triggers_cert_rotation(self):
             assert resp == mock_resp_200
             mock_check.assert_called_once()
             mock_conf.assert_called_once()
-            cb = (
-                mock_conf.call_args.args[0]
-                if mock_conf.call_args.args
-                else mock_conf.call_args.kwargs["client_cert_callback"]
+            # The rotation path passes the rotated cert/key explicitly rather
+            # than temporarily swapping the shared `_client_cert_callback`,
+            # which must therefore be left untouched.
+            assert mock_conf.call_args.kwargs["_cert_key_override"] == (
+                new_cert,
+                new_key,
             )
-            assert cb() == (new_cert, new_key)
+            assert session._client_cert_callback is None
 
         await session.close()
 
@@ -1008,7 +1012,7 @@ async def dummy_completed():
         new_cert = b"new_cert"
         new_key = b"new_key"
 
-        async def fake_configure(cb=None):
+        async def fake_configure(cb=None, **kwargs):
             session._mtls_init_task = asyncio.create_task(dummy_completed())
 
         with (
@@ -1613,60 +1617,81 @@ async def test_cert_rotation_credential_refresh_not_implemented_retries(self):
 
     @pytest.mark.asyncio
     async def test_401_mtls_consecutive_multi_rotation(self):
-        """Verifies that multiple consecutive rotations (v1 -> v2 -> v3) succeed."""
+        """Verifies that consecutive rotations (v1 -> v2 -> v3) succeed.
+
+        `configure_mtls_channel` is deliberately NOT mocked here. The
+        low-level helpers are patched instead so the real reconfiguration path
+        executes; otherwise this test would still pass even if rotation
+        stopped swapping the transport or started clobbering the shared
+        `_client_cert_callback`.
+        """
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
         mock_creds.before_request = mock.AsyncMock(return_value=None)
         mock_creds.refresh = mock.AsyncMock(return_value=None)
 
-        session = sessions.AsyncAuthorizedSession(mock_creds)
-        session._is_mtls = True
-        session._cached_cert = b"cert_v1"
-
-        # Rotation 1: v1 -> v2
-        with mock.patch(
-            "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
-            new_callable=mock.AsyncMock,
-            return_value=(b"cert_v2", b"key_v2", b"fp1", b"fp2"),
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert_v1", b"key_v1"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
         ):
-            with mock.patch.object(
-                session, "configure_mtls_channel", new_callable=mock.AsyncMock
-            ) as mock_conf:
-                mock_auth = mock.AsyncMock(
-                    side_effect=[
-                        mock.Mock(status_code=401, close=mock.AsyncMock()),
-                        mock.Mock(status_code=200, close=mock.AsyncMock()),
-                    ]
-                )
-                session._auth_request = mock_auth
-                await session.request("GET", "https://pubsub.mtls.googleapis.com/test")
-                mock_conf.assert_called_once()
-                assert session._client_cert_callback is None
 
-        session._cached_cert = b"cert_v2"
+            def user_cb():
+                return (b"cert_v1", b"key_v1")
 
-        # Rotation 2: v2 -> v3 (Must still have client_cert_callback == None to read disk)
-        with mock.patch(
-            "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response",
-            new_callable=mock.AsyncMock,
-            return_value=(b"cert_v3", b"key_v3", b"fp2", b"fp3"),
-        ) as mock_check:
-            with mock.patch.object(
-                session, "configure_mtls_channel", new_callable=mock.AsyncMock
-            ) as mock_conf:
-                mock_auth = mock.AsyncMock(
-                    side_effect=[
-                        mock.Mock(status_code=401, close=mock.AsyncMock()),
-                        mock.Mock(status_code=200, close=mock.AsyncMock()),
-                    ]
-                )
-                session._auth_request = mock_auth
-                await session.request("GET", "https://pubsub.mtls.googleapis.com/test")
-                # Verify check was called with callback=None (allowing disk read)
-                mock_check.assert_called_with(b"cert_v2", None)
-                mock_conf.assert_called_once()
-                assert session._client_cert_callback is None
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            # Configure with an explicit, non-None user callback. A rotation
+            # that clobbers this shared attribute is then observable; with a
+            # default of None the overwrite would be a silent no-op.
+            await session.configure_mtls_channel(user_cb)
+            assert session._cached_cert == b"cert_v1"
+            assert session._client_cert_callback is user_cb
+
+            rotations = ((b"cert_v1", b"cert_v2"), (b"cert_v2", b"cert_v3"))
+            for old_cert, new_cert in rotations:
+                prev_auth_request = session._auth_request
+                with (
+                    mock.patch(
+                        "google.auth.aio.transport.mtls."
+                        "check_parameters_for_unauthorized_response",
+                        new_callable=mock.AsyncMock,
+                        return_value=(new_cert, b"key", b"fp_old", b"fp_new"),
+                    ) as mock_check,
+                    mock.patch.object(
+                        sessions.AiohttpRequest,
+                        "__call__",
+                        side_effect=[
+                            mock.Mock(status_code=401, close=mock.AsyncMock()),
+                            mock.Mock(status_code=200, close=mock.AsyncMock()),
+                        ],
+                    ),
+                ):
+                    resp = await session.request(
+                        "GET", "https://pubsub.mtls.googleapis.com/test"
+                    )
 
-        await session.close()
+                assert resp.status_code == 200
+                # The check is driven by the previously cached cert and the
+                # user's callback (None), so the on-disk cert can be read.
+                mock_check.assert_called_once_with(old_cert, user_cb)
+                # Real reconfiguration ran: cert cached and transport swapped.
+                assert session._cached_cert == new_cert
+                assert session._auth_request is not prev_auth_request
+                assert session.is_mtls is True
+                # Shared callback state must survive rotation untouched.
+                assert session._client_cert_callback is user_cb
+
+            await session.close()
 
     @pytest.mark.asyncio
     async def test_non_mtls_not_implemented_refresh_returns_401_without_retry(self):
@@ -1803,3 +1828,214 @@ async def test_configure_mtls_channel_force_reconfigures_same_callback(self):
             await session.configure_mtls_channel(force=True)
             assert session._mtls_init_task is not task1
             await session.close()
+
+    @pytest.mark.asyncio
+    async def test_concurrent_rotation_retries_for_non_refreshable_credentials(self):
+        """Concurrent 401s with non-refreshable credentials must both retry.
+
+        Only one coroutine performs the rotation; the other skips the check
+        block via the dedupe counter. The skipping coroutine must still observe
+        that the channel was reconfigured since its own 401 and retry, rather
+        than returning the stale 401 to the caller.
+        """
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(side_effect=NotImplementedError)
+
+        def _resp(status):
+            return mock.Mock(status_code=status, close=mock.AsyncMock())
+
+        mock_resp_401_1 = _resp(http_client.UNAUTHORIZED)
+        mock_resp_401_2 = _resp(http_client.UNAUTHORIZED)
+        mock_resp_200_1 = _resp(http_client.OK)
+        mock_resp_200_2 = _resp(http_client.OK)
+
+        mock_auth_req = mock.AsyncMock(
+            side_effect=[
+                mock_resp_401_1,
+                mock_resp_401_2,
+                mock_resp_200_1,
+                mock_resp_200_2,
+            ]
+        )
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        async def slow_check(*args, **kwargs):
+            # Hold the rotation lock long enough that the second coroutine
+            # queues behind it and then takes the dedupe path.
+            await asyncio.sleep(0.05)
+            return (b"new_cert", b"new_key", b"old_fp", b"new_fp")
+
+        with (
+            mock.patch(
+                "google.auth.aio.transport.mtls."
+                "check_parameters_for_unauthorized_response",
+                side_effect=slow_check,
+            ) as mock_check,
+            mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf,
+        ):
+            results = await asyncio.gather(
+                session.request("GET", "https://pubsub.mtls.googleapis.com/t1"),
+                session.request("GET", "https://pubsub.mtls.googleapis.com/t2"),
+            )
+
+        # Exactly one coroutine ran the check and the rotation.
+        assert mock_check.call_count == 1
+        assert mock_conf.call_count == 1
+        # Both requests must have been retried on the rotated channel.
+        assert results == [mock_resp_200_1, mock_resp_200_2]
+        assert mock_auth_req.call_count == 4
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_reconfigure_to_non_mtls_replaces_stale_mtls_transport(self):
+        """A session leaving mTLS must not keep serving the old client cert."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            with mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ):
+                session = sessions.AsyncAuthorizedSession(mock_creds)
+                await session.configure_mtls_channel()
+
+            assert session.is_mtls is True
+            mtls_transport = session._auth_request
+
+            # The workload stops providing a client certificate.
+            with mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(False, None, None),
+            ):
+                await session.configure_mtls_channel(force=True)
+
+            assert session.is_mtls is False
+            assert session._cached_cert is None
+            # The stale mTLS transport must be retired, not silently reused.
+            assert session._auth_request is not mtls_transport
+            assert mtls_transport in session._old_auth_requests
+
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_propagates_caller_cancellation(self):
+        """Cancelling the caller must raise, not be swallowed by cleanup."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+
+        async def slow_get_cert(cb=None):
+            await asyncio.sleep(10)
+            return (True, b"cert", b"key")
+
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                side_effect=slow_get_cert,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            caller = asyncio.create_task(session.configure_mtls_channel())
+            await asyncio.sleep(0.02)
+
+            caller.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await caller
+
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_follows_replacement_task(self):
+        """A caller awaiting a task that gets replaced follows the new one."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+
+        release = asyncio.Event()
+
+        async def gated_get_cert(cb=None):
+            await release.wait()
+            return (True, b"cert", b"key")
+
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                side_effect=gated_get_cert,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession"),
+        ):
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            # First caller starts and blocks on the gated configuration.
+            waiter = asyncio.create_task(session.configure_mtls_channel())
+            await asyncio.sleep(0.02)
+            first_task = session._mtls_init_task
+            assert first_task is not None
+
+            # A forced reconfiguration cancels and replaces that task.
+            release.set()
+            await session.configure_mtls_channel(force=True)
+            assert session._mtls_init_task is not first_task
+
+            # The original caller must not surface a cancellation it never
+            # requested; it follows the replacement task instead.
+            await waiter
+            assert session.is_mtls is True
+
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_retrieve_task_exception_helper(self):
+        """The done-callback marks failures retrieved and tolerates cancels."""
+
+        async def boom():
+            raise RuntimeError("boom")
+
+        task = asyncio.create_task(boom())
+        task.add_done_callback(sessions._retrieve_task_exception)
+        with pytest.raises(RuntimeError):
+            await task
+
+        async def sleeper():
+            await asyncio.sleep(10)
+
+        cancelled = asyncio.create_task(sleeper())
+        cancelled.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await cancelled
+        # Must not raise CancelledError/InvalidStateError when inspected.
+        sessions._retrieve_task_exception(cancelled)

From 7736b55e10fec31581030f9444b18b59a6e193ed Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sun, 13 Sep 2026 05:53:54 +0000
Subject: [PATCH 20/22] fix(auth): apply non-swallowing cancellation handling
 to close()

`close()` still used `cancel()` + `await task` + `except (Exception,
asyncio.CancelledError): pass`, the same anti-pattern flagged in review on
`configure_mtls_channel`. Replaced with `asyncio.wait()` so a cancellation
aimed at `close()` is no longer absorbed.

TAG=agy
CONV=547fc656-aa8a-4bda-83c8-d57057515ec3
---
 .../google-auth/google/auth/aio/transport/sessions.py     | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 54d6394e0b9c..f9992b949af8 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -982,10 +982,10 @@ async def close(self) -> None:
         try:
             if self._mtls_init_task and not self._mtls_init_task.done():
                 self._mtls_init_task.cancel()
-                try:
-                    await self._mtls_init_task
-                except (Exception, asyncio.CancelledError):
-                    pass
+                # Same rationale as `configure_mtls_channel`: `asyncio.wait`
+                # lets the cancelled initialization task unwind without
+                # absorbing a cancellation aimed at this `close()` call.
+                await asyncio.wait({self._mtls_init_task})
         finally:
             while self._old_auth_requests:
                 old_request = self._old_auth_requests.pop(0)

From ab3fe1d658f5302bbfe6a7e5a0672c4e50aff731 Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sun, 13 Sep 2026 06:41:17 +0000
Subject: [PATCH 21/22] fix(auth): correct post-rotation refresh and
 close/rotation race in async mTLS

A mutation sweep over this PR (revert each fix, check whether any test fails)
found that 6 of 14 behaviours had no test protection at all, including the
`asyncio.wait` change from review and the whole `close()` fix. Closing those
gaps surfaced two real bugs.

1. A rotation no longer guarantees a post-rotation credential refresh.
   `_refresh_counter_at_last_reconfig` counted refresh *completions*, so a
   refresh that started before a rotation and finished after it was credited
   as post-rotation. The rotating coroutine then skipped its own refresh and
   retried with a certificate-bound token minted over the old transport.
   Replaced with `_last_refresh_reconfig_gen`, which records the
   reconfiguration generation observed when a refresh *starts*.

2. `close()` raced with an in-flight rotation and leaked a transport. The
   rotation installed a fresh `aiohttp.ClientSession` after `close()` had
   already drained everything, resurrecting the session and stranding an open
   client session. Added a `_closed` flag, set under `_mtls_init_lock`, that
   makes `configure_mtls_channel()` refuse to build new state.

Also in this change:
- `check_use_client_cert` now runs inside the try block, so a failure reading
  the cert config surfaces as the documented `MutualTLSChannelError`.
- The five silent `except Exception: pass` cleanup paths now log at debug.
- Documented that a failed rotation deliberately leaves `_mtls_check_counter`
  un-incremented so a queued coroutine retries rather than inheriting it.

Adds 8 regression tests. Every one was confirmed to fail with its fix reverted;
the sweep now reports 18 killed / 0 survived.

TAG=agy
CONV=547fc656-aa8a-4bda-83c8-d57057515ec3
---
 .../google/auth/aio/transport/sessions.py     | 130 ++++++--
 .../tests/transport/aio/test_sessions_mtls.py | 313 ++++++++++++++++++
 2 files changed, 407 insertions(+), 36 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index f9992b949af8..1c91892bbc7f 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -183,17 +183,23 @@ def __init__(
         # rotation check (because a concurrent request already performed it)
         # still observe that the channel changed since its own 401.
         self._mtls_reconfig_counter = 0
-        # Value of `_refresh_counter` at the moment of the most recent
-        # reconfiguration. A credential refresh is only considered redundant
-        # if it happened *after* the latest channel change, since a
-        # certificate-bound token minted on the old channel is not valid on
-        # the new one.
-        self._refresh_counter_at_last_reconfig = 0
+        # Value of `_mtls_reconfig_counter` observed when the most recently
+        # completed credential refresh *started*. Counting refresh completions
+        # is not sufficient to decide whether a token is usable on the current
+        # channel: a refresh that began before a rotation and finished after it
+        # was still minted over the old transport, and a certificate-bound
+        # token from the old channel is rejected by the new one. Recording the
+        # generation a refresh started in lets us tell the two apart.
+        self._last_refresh_reconfig_gen = -1
         # Serializes the decision to create a new mTLS initialization task so
         # that two concurrent callers cannot both spawn `_do_configure()`.
         self._mtls_init_lock: Optional[asyncio.Lock] = None
         self._refresh_lock: Optional[asyncio.Lock] = None
         self._refresh_counter = 0
+        # Set by `close()`. Guarded by `_mtls_init_lock` so that an in-flight
+        # certificate rotation cannot install a new transport on a session that
+        # has already been torn down.
+        self._closed = False
 
     async def _trim_old_auth_requests(self) -> None:
         """Close retired transports, keeping at most the two most recent."""
@@ -204,8 +210,10 @@ async def _trim_old_auth_requests(self) -> None:
                     res = oldest_auth_request.close()
                     if inspect.isawaitable(res):
                         await res
-            except Exception:
-                pass
+            except Exception as caught_exc:
+                _LOGGER.debug(
+                    "Failed to close a retired auth transport: %s", caught_exc
+                )
 
     async def _reset_non_mtls_state(self) -> None:
         """Clear mTLS state, retiring a library-owned mTLS transport.
@@ -268,6 +276,8 @@ async def configure_mtls_channel(
         Raises:
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
+            google.auth.exceptions.InvalidOperation: If the session has already
+                been closed.
         """
         if self._mtls_init_lock is None:
             self._mtls_init_lock = asyncio.Lock()
@@ -277,6 +287,19 @@ async def configure_mtls_channel(
         # both spawn `_do_configure()`. The lock is released before awaiting the
         # task itself, so a slow configuration does not block unrelated callers.
         async with init_lock:
+            if self._closed:
+                # Without this, a rotation triggered by an in-flight request
+                # could build and install a brand new transport after `close()`
+                # has already drained everything, leaking it permanently.
+                #
+                # A task that was *already* running when `close()` landed needs
+                # no separate check: `close()` sets `_closed` and cancels the
+                # task without yielding in between, so a running
+                # `_do_configure` is always interrupted at one of its awaits
+                # before it reaches the point where it installs a transport.
+                raise exceptions.InvalidOperation(
+                    "Cannot configure the mTLS channel on a closed session."
+                )
             if _cert_key_override is not None:
                 needs_reconfig = True
             else:
@@ -315,10 +338,17 @@ async def configure_mtls_channel(
                     self._client_cert_callback = client_cert_callback
 
                 async def _do_configure():
-                    # Run the blocking check in an executor
-                    use_client_cert = await mtls._run_in_executor(
-                        google.auth.transport._mtls_helper.check_use_client_cert
-                    )
+                    # Run the blocking check in an executor. It reads and parses
+                    # a config file, so it can fail in ways the caller is
+                    # promised to see as `MutualTLSChannelError`.
+                    try:
+                        use_client_cert = await mtls._run_in_executor(
+                            google.auth.transport._mtls_helper.check_use_client_cert
+                        )
+                    except Exception as caught_exc:
+                        raise exceptions.MutualTLSChannelError(
+                            caught_exc
+                        ) from caught_exc
                     if not use_client_cert:
                         return
 
@@ -580,11 +610,15 @@ async def _recover_auth_state():
                                                             call_key_bytes,
                                                         )
                                                     )
-                                                    self._refresh_counter_at_last_reconfig = (
-                                                        self._refresh_counter
-                                                    )
                                                     self._mtls_reconfig_counter += 1
                                                 except Exception as e:
+                                                    # NOTE: `_mtls_check_counter`
+                                                    # is deliberately left
+                                                    # un-incremented below, so a
+                                                    # queued coroutine retries
+                                                    # the reconfiguration rather
+                                                    # than inheriting this
+                                                    # failure.
                                                     _LOGGER.error(
                                                         "Failed to reconfigure mTLS channel: %s",
                                                         e,
@@ -617,21 +651,27 @@ async def _recover_auth_state():
                             self._refresh_lock = asyncio.Lock()
 
                         async with self._refresh_lock:
-                            # A refresh is redundant only if it happened after
-                            # both this request's 401 and the most recent
-                            # channel reconfiguration. Using the later of the
-                            # two keeps concurrent requests de-duplicated while
-                            # still guaranteeing at least one refresh against a
-                            # freshly rotated channel.
-                            refresh_baseline = max(
-                                refresh_counter_at_error,
-                                self._refresh_counter_at_last_reconfig,
+                            # A concurrent refresh only makes this one redundant
+                            # if it completed after this request's 401 *and* it
+                            # was started on the channel we are about to retry
+                            # on. Completion order alone is not enough: a
+                            # refresh that began before a rotation and finished
+                            # after it minted its token over the old transport,
+                            # and the rotated channel will reject it.
+                            already_refreshed = (
+                                self._refresh_counter > refresh_counter_at_error
+                                and self._last_refresh_reconfig_gen
+                                >= self._mtls_reconfig_counter
                             )
-                            if self._refresh_counter > refresh_baseline:
+                            if already_refreshed:
                                 _LOGGER.debug(
                                     "Credentials were already refreshed by a concurrent task. Skipping duplicate refresh."
                                 )
                             else:
+                                # Snapshot the generation *before* awaiting so a
+                                # rotation that lands mid-refresh is not
+                                # credited to the token we are about to mint.
+                                reconfig_gen = self._mtls_reconfig_counter
                                 try:
                                     await self._credentials.refresh(self._auth_request)
                                 except NotImplementedError:
@@ -655,6 +695,7 @@ async def _recover_auth_state():
                                     return response
                                 else:
                                     self._refresh_counter += 1
+                                    self._last_refresh_reconfig_gen = reconfig_gen
 
                         if is_streaming:
                             return response
@@ -671,8 +712,10 @@ async def _recover_auth_state():
                             res = response.close()
                             if inspect.isawaitable(res):
                                 await res
-                        except Exception:
-                            pass
+                        except Exception as close_exc:
+                            _LOGGER.debug(
+                                "Failed to close the 401 response: %s", close_exc
+                            )
                     raise
                 # If it returned a response (meaning streaming or error), bail out
                 if early_return_response is not None:
@@ -682,8 +725,8 @@ async def _recover_auth_state():
                         res = response.close()
                         if inspect.isawaitable(res):
                             await res
-                    except Exception:
-                        pass
+                    except Exception as close_exc:
+                        _LOGGER.debug("Failed to close the 401 response: %s", close_exc)
                 if max_allowed_time is not None:
                     remaining_time = max(
                         0.0, max_allowed_time - (time.monotonic() - start_time)
@@ -978,14 +1021,25 @@ def is_mtls(self):
     async def close(self) -> None:
         """
         Close the underlying auth request session.
+
+        Once closed, the session refuses further mTLS (re)configuration, so an
+        in-flight certificate rotation cannot resurrect it with a freshly built
+        transport that nothing would ever close.
         """
+        if self._mtls_init_lock is None:
+            self._mtls_init_lock = asyncio.Lock()
+        # Flip the flag under the same lock `configure_mtls_channel` uses to
+        # decide whether to spawn a task, so the two cannot interleave.
+        async with self._mtls_init_lock:
+            self._closed = True
+            init_task = self._mtls_init_task
         try:
-            if self._mtls_init_task and not self._mtls_init_task.done():
-                self._mtls_init_task.cancel()
+            if init_task and not init_task.done():
+                init_task.cancel()
                 # Same rationale as `configure_mtls_channel`: `asyncio.wait`
                 # lets the cancelled initialization task unwind without
                 # absorbing a cancellation aimed at this `close()` call.
-                await asyncio.wait({self._mtls_init_task})
+                await asyncio.wait({init_task})
         finally:
             while self._old_auth_requests:
                 old_request = self._old_auth_requests.pop(0)
@@ -994,12 +1048,16 @@ async def close(self) -> None:
                         res = old_request.close()
                         if inspect.isawaitable(res):
                             await res
-                except Exception:
-                    pass
+                except Exception as caught_exc:
+                    _LOGGER.debug(
+                        "Failed to close a retired auth transport: %s", caught_exc
+                    )
             try:
                 if hasattr(self._auth_request, "close"):
                     res = self._auth_request.close()
                     if inspect.isawaitable(res):
                         await res
-            except Exception:
-                pass
+            except Exception as caught_exc:
+                _LOGGER.debug(
+                    "Failed to close the active auth transport: %s", caught_exc
+                )
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index aebbfae40ec8..8a7e9905bb38 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -2039,3 +2039,316 @@ async def sleeper():
             await cancelled
         # Must not raise CancelledError/InvalidStateError when inspected.
         sessions._retrieve_task_exception(cancelled)
+
+    @pytest.mark.asyncio
+    async def test_reconfigure_cancellation_while_retiring_old_task_propagates(self):
+        """A cancellation aimed at the reconfiguring coroutine must not be eaten.
+
+        `configure_mtls_channel` cancels the task it is replacing and waits for
+        it to unwind. That wait has to absorb only the *retired task's*
+        cancellation; a cancellation targeting the caller still has to
+        propagate.
+        """
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+
+        calls = {"n": 0}
+
+        async def run_in_executor(*args, **kwargs):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                # Take a measurable amount of time to unwind so the replacing
+                # coroutine is still parked in the wait when we cancel it.
+                try:
+                    await asyncio.sleep(10)
+                except asyncio.CancelledError:
+                    await asyncio.sleep(0.05)
+                    raise
+            # Any replacement task finishes immediately.
+            return False
+
+        with mock.patch.object(sessions.mtls, "_run_in_executor", new=run_in_executor):
+            first = asyncio.create_task(session.configure_mtls_channel())
+            await asyncio.sleep(0.01)
+            assert session._mtls_init_task is not None
+
+            replacer = asyncio.create_task(session.configure_mtls_channel(force=True))
+            await asyncio.sleep(0.01)  # parked waiting on the retired task
+
+            replacer.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await replacer
+
+            first.cancel()
+            try:
+                await first
+            except (asyncio.CancelledError, exceptions.MutualTLSChannelError):
+                pass
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_close_cancels_in_flight_mtls_init(self):
+        """`close()` must actually cancel a still-running initialization task."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+
+        async def never(*args, **kwargs):
+            await asyncio.sleep(10)
+
+        with mock.patch.object(sessions.mtls, "_run_in_executor", new=never):
+            waiter = asyncio.create_task(session.configure_mtls_channel())
+            await asyncio.sleep(0.01)
+            init_task = session._mtls_init_task
+            assert init_task is not None
+            assert not init_task.done()
+
+            await session.close()
+
+            assert init_task.done()
+            assert init_task.cancelled()
+
+            with pytest.raises(asyncio.CancelledError):
+                await waiter
+
+    @pytest.mark.asyncio
+    async def test_close_propagates_its_own_cancellation(self):
+        """A cancellation aimed at `close()` must not be absorbed."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+
+        async def stubborn(*args, **kwargs):
+            try:
+                await asyncio.sleep(10)
+            except asyncio.CancelledError:
+                # Unwind slowly so `close()` is parked in the wait.
+                await asyncio.sleep(0.05)
+                raise
+
+        with mock.patch.object(sessions.mtls, "_run_in_executor", new=stubborn):
+            waiter = asyncio.create_task(session.configure_mtls_channel())
+            await asyncio.sleep(0.01)
+
+            closing = asyncio.create_task(session.close())
+            await asyncio.sleep(0.01)
+            closing.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await closing
+
+            waiter.cancel()
+            try:
+                await waiter
+            except (asyncio.CancelledError, exceptions.MutualTLSChannelError):
+                pass
+
+    @pytest.mark.asyncio
+    async def test_failed_mtls_init_without_awaiter_is_marked_retrieved(self):
+        """A background init failure nobody awaits must not warn at GC time."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+
+        async def slow_boom(*args, **kwargs):
+            await asyncio.sleep(0.05)
+            raise RuntimeError("cert check exploded")
+
+        with mock.patch.object(sessions.mtls, "_run_in_executor", new=slow_boom):
+            # The only caller gives up before the task fails.
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(session.configure_mtls_channel(), 0.01)
+            task = session._mtls_init_task
+            assert task is not None
+            await asyncio.wait({task})
+
+        assert task.done()
+        assert not task.cancelled()
+        # The done-callback must have consumed the exception already. Without
+        # it, CPython would still have the task flagged for the
+        # "Task exception was never retrieved" report at collection time.
+        #
+        # This has to be checked BEFORE calling `task.exception()` below, since
+        # retrieving the exception here would clear the flag by itself and make
+        # the assertion vacuous.
+        assert task._log_traceback is False
+        assert isinstance(task.exception(), exceptions.MutualTLSChannelError)
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_rotation_forces_refresh_when_earlier_refresh_lands_late(self):
+        """A refresh straddling a rotation must not satisfy the post-rotation one.
+
+        A refresh that starts before the channel is rotated mints its token
+        over the old transport. Even though it completes after the rotation,
+        the rotating coroutine still has to perform its own refresh, otherwise
+        it retries carrying a token the new channel will reject.
+        """
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+
+        async def slow_refresh(_transport):
+            await asyncio.sleep(0.10)
+
+        mock_creds.refresh = mock.AsyncMock(side_effect=slow_refresh)
+
+        seen = {}
+
+        async def auth_req(url, *args, **kwargs):
+            seen[url] = seen.get(url, 0) + 1
+            status = http_client.UNAUTHORIZED if seen[url] == 1 else http_client.OK
+            return mock.Mock(status_code=status, close=mock.AsyncMock())
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock(side_effect=auth_req)
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        async def fast_check(*args, **kwargs):
+            await asyncio.sleep(0.01)
+            return (b"new_cert", b"new_key", b"old_fp", b"new_fp")
+
+        with (
+            mock.patch(
+                "google.auth.aio.transport.mtls."
+                "check_parameters_for_unauthorized_response",
+                side_effect=fast_check,
+            ),
+            mock.patch.object(
+                session, "configure_mtls_channel", new_callable=mock.AsyncMock
+            ) as mock_conf,
+        ):
+            # The non-mTLS request takes `_refresh_lock` before the rotation
+            # begins and releases it only after the rotation has finished.
+            plain = asyncio.create_task(
+                session.request("GET", "https://example.com/plain")
+            )
+            await asyncio.sleep(0)
+            rotating = asyncio.create_task(
+                session.request("GET", "https://pubsub.mtls.googleapis.com/x")
+            )
+            await asyncio.gather(plain, rotating)
+
+        assert mock_conf.call_count == 1
+        # One refresh from the plain request (old channel) plus one forced by
+        # the rotation. Counting completions alone would wrongly treat the
+        # first as satisfying the second.
+        assert mock_creds.refresh.call_count == 2
+        assert session._last_refresh_reconfig_gen == session._mtls_reconfig_counter
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_rejects_a_closed_session(self):
+        """A closed session must refuse to build new mTLS state."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+        await session.close()
+
+        with pytest.raises(exceptions.InvalidOperation):
+            await session.configure_mtls_channel()
+        assert session._mtls_init_task is None
+
+    @pytest.mark.asyncio
+    async def test_rotation_racing_close_does_not_leak_a_transport(self):
+        """An in-flight rotation must not install a transport after `close()`.
+
+        Otherwise `close()` returns, the rotation then builds a fresh
+        `aiohttp.ClientSession` and installs it on the dead session, and
+        nothing ever closes it.
+        """
+        created = []
+
+        class _FakeClientSession:
+            def __init__(self, *args, **kwargs):
+                self.closed = False
+                created.append(self)
+
+            async def close(self):
+                self.closed = True
+
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        seen = {}
+
+        async def auth_req(url, *args, **kwargs):
+            seen[url] = seen.get(url, 0) + 1
+            status = http_client.UNAUTHORIZED if seen[url] == 1 else http_client.OK
+            return mock.Mock(status_code=status, close=mock.AsyncMock())
+
+        gate = asyncio.Event()
+
+        async def blocking_check(*args, **kwargs):
+            await gate.wait()
+            return (b"new_cert", b"new_key", b"old_fp", b"new_fp")
+
+        with (
+            mock.patch(
+                "google.auth.transport._mtls_helper.check_use_client_cert",
+                return_value=True,
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls.make_client_cert_ssl_context",
+                return_value=mock.Mock(spec=ssl.SSLContext),
+            ),
+            mock.patch("aiohttp.TCPConnector"),
+            mock.patch("aiohttp.ClientSession", _FakeClientSession),
+            mock.patch(
+                "google.auth.aio.transport.mtls.get_client_cert_and_key",
+                return_value=(True, b"cert", b"key"),
+            ),
+            mock.patch(
+                "google.auth.aio.transport.mtls."
+                "check_parameters_for_unauthorized_response",
+                side_effect=blocking_check,
+            ),
+            mock.patch.object(
+                sessions.AiohttpRequest, "__call__", side_effect=auth_req
+            ),
+        ):
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+            await session.configure_mtls_channel()
+            assert len(created) == 1
+
+            pending = asyncio.create_task(
+                session.request("GET", "https://pubsub.mtls.googleapis.com/x")
+            )
+            await asyncio.sleep(0.02)  # park inside the rotation
+
+            await session.close()
+
+            gate.set()
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                await pending
+
+        # No second transport was built, and the original one was closed.
+        assert len(created) == 1
+        assert created[0].closed is True
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_wraps_use_client_cert_failure(self):
+        """A failure reading the cert config must surface as MutualTLSChannelError."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_use_client_cert",
+            side_effect=UnicodeDecodeError("utf-8", b"\xff", 0, 1, "bad byte"),
+        ):
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                await session.configure_mtls_channel()
+
+        await session.close()

From 74852e63b2f31c90a08b5a93d7d4705a22c39b03 Mon Sep 17 00:00:00 2001
From: Andy Zhao <andyzhao@google.com>
Date: Sun, 13 Sep 2026 07:58:12 +0000
Subject: [PATCH 22/22] fix(auth): re-read rotation state at point of use in
 401 recovery

A third review pass over this PR found that the fix for the reviewer's first
finding was incomplete. `channel_reconfigured` was correctly moved off a
coroutine-local flag and onto the session-level `_mtls_reconfig_counter`, but
it was still evaluated once, immediately after the rotation block, and only
consumed much later inside the refresh exception handlers -- after awaiting
`_refresh_lock` and after the refresh itself.

A rotation performed by a concurrent coroutine in that window was therefore
invisible, and a request whose credentials cannot be refreshed returned its
stale 401 instead of retrying on the freshly rotated channel. That is the same
user-visible symptom the original finding described, and it contradicts the
documented behaviour that static or certificate-bound credentials still retry
when a rotation has reconfigured the channel.

Reproduced deterministically: coroutine A evaluates the flag as False, parks
inside a failing refresh, coroutine B rotates the channel, and A then returns
401 while B returns 200.

`channel_reconfigured` becomes a closure evaluated at each point of use, so the
decision reads the counter at the moment it is made rather than two awaits
earlier.

Adds `test_rotation_landing_during_refresh_still_triggers_retry`, confirmed to
fail against the unfixed source. The mutation sweep gains a case that restores
the snapshot semantics and now reports 19 killed / 0 survived.

TAG=agy
CONV=547fc656-aa8a-4bda-83c8-d57057515ec3
---
 .../google/auth/aio/transport/sessions.py     | 19 +++--
 .../tests/transport/aio/test_sessions_mtls.py | 84 +++++++++++++++++++
 2 files changed, 98 insertions(+), 5 deletions(-)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index 1c91892bbc7f..0763ea4cbb34 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -639,13 +639,22 @@ async def _recover_auth_state():
                                                     )
                                         # Always increment so waiting tasks skip the check block
                                         self._mtls_check_counter += 1
+
                         # Derived from session state rather than a local flag:
                         # a concurrent request may have performed the rotation
                         # on our behalf (we then skipped the check block), and
                         # that request still needs to retry on the new channel.
-                        channel_reconfigured = (
-                            self._mtls_reconfig_counter > reconfig_counter_at_error
-                        )
+                        #
+                        # Evaluated at each use rather than snapshotted here: a
+                        # concurrent request can rotate the channel while this
+                        # coroutine is queued on `_refresh_lock` or waiting for
+                        # its own refresh to fail. A value captured at this
+                        # point would miss that rotation and drop a retry that
+                        # would have succeeded on the new channel.
+                        def channel_reconfigured() -> bool:
+                            return (
+                                self._mtls_reconfig_counter > reconfig_counter_at_error
+                            )
 
                         if self._refresh_lock is None:
                             self._refresh_lock = asyncio.Lock()
@@ -678,14 +687,14 @@ async def _recover_auth_state():
                                     _LOGGER.debug(
                                         "Credentials do not implement refresh()."
                                     )
-                                    if not channel_reconfigured:
+                                    if not channel_reconfigured():
                                         return response
                                 except exceptions.InvalidOperation as e:
                                     _LOGGER.debug(
                                         "Credentials cannot be refreshed: %s",
                                         e,
                                     )
-                                    if not channel_reconfigured:
+                                    if not channel_reconfigured():
                                         return response
                                 except exceptions.RefreshError as e:
                                     _LOGGER.debug(
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index 8a7e9905bb38..a0f313fe5899 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -2352,3 +2352,87 @@ async def test_configure_mtls_channel_wraps_use_client_cert_failure(self):
                 await session.configure_mtls_channel()
 
         await session.close()
+
+    @pytest.mark.asyncio
+    async def test_rotation_landing_during_refresh_still_triggers_retry(self):
+        """A rotation that lands while we await refresh must not be missed.
+
+        Whether the channel moved since this request's 401 has to be read at
+        the point the decision is made, not snapshotted before acquiring
+        `_refresh_lock`. A coroutine whose credentials cannot be refreshed
+        would otherwise return its stale 401 even though a concurrent
+        coroutine rotated the channel in the meantime.
+        """
+        a_in_refresh = asyncio.Event()
+        b_rotated = asyncio.Event()
+
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+
+        first_refresh = {"done": False}
+
+        async def refresh_side_effect(*args, **kwargs):
+            if not first_refresh["done"]:
+                first_refresh["done"] = True
+                a_in_refresh.set()
+                await b_rotated.wait()
+            raise exceptions.InvalidOperation("static cert-bound token")
+
+        mock_creds.refresh = mock.AsyncMock(side_effect=refresh_side_effect)
+
+        seen = set()
+
+        async def auth_request(url, method, data, headers, timeout, **kwargs):
+            if url not in seen:
+                seen.add(url)
+                return mock.Mock(
+                    status_code=http_client.UNAUTHORIZED, close=mock.AsyncMock()
+                )
+            return mock.Mock(status_code=http_client.OK, close=mock.AsyncMock())
+
+        mock_auth_req = mock.AsyncMock(side_effect=auth_request)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        checks = {"n": 0}
+
+        async def check_side_effect(cached_cert, callback):
+            checks["n"] += 1
+            if checks["n"] == 1:
+                # The first coroutine sees an unchanged certificate.
+                return (b"old_cert", b"old_key", b"same_fp", b"same_fp")
+            return (b"new_cert", b"new_key", b"old_fp", b"new_fp")
+
+        async def fake_configure(*args, **kwargs):
+            b_rotated.set()
+
+        with (
+            mock.patch(
+                "google.auth.aio.transport.mtls."
+                "check_parameters_for_unauthorized_response",
+                side_effect=check_side_effect,
+            ),
+            mock.patch.object(
+                session, "configure_mtls_channel", side_effect=fake_configure
+            ),
+        ):
+            task_a = asyncio.create_task(
+                session.request("GET", "https://pubsub.mtls.googleapis.com/a")
+            )
+            await asyncio.wait_for(a_in_refresh.wait(), 5)
+            task_b = asyncio.create_task(
+                session.request("GET", "https://pubsub.mtls.googleapis.com/b")
+            )
+            resp_a, resp_b = await asyncio.wait_for(asyncio.gather(task_a, task_b), 5)
+
+        assert session._mtls_reconfig_counter == 1
+        # The rotating coroutine retries, and so must the one that only
+        # learned about the rotation after its own refresh failed.
+        assert resp_b.status_code == http_client.OK
+        assert resp_a.status_code == http_client.OK
+
+        await session.close()