fix(auth): avoid creating mTLS SSL context for custom async transports

packages/google-auth/google/auth/aio/transport/sessions.py

@@ -192,14 +192,13 @@ async def _do_configure():
                     ) = await mtls.get_client_cert_and_key(client_cert_callback)
 
                     if is_mtls:
-                        ssl_context = await mtls._run_in_executor(
-                            mtls.make_client_cert_ssl_context, cert, key
-                        )
-
                         # 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)
 

packages/google-auth/tests/transport/aio/test_sessions.py

@@ -335,6 +335,32 @@ async def test_http_delete_method_success(self):
             assert await response.read() == expected_payload
             response = await authed_session.close()
 
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_with_custom_transport_and_broken_cert(self):
+        auth_request = MockRequest()
+        authed_session = sessions.AsyncAuthorizedSession(
+            self.credentials, auth_request=auth_request
+        )
+
+        with patch(
+            "google.auth.transport._mtls_helper.check_use_client_cert",
+            return_value=True,
+        ):
+
+            def callback():
+                return b"invalid-cert", b"invalid-key"
+
+            with pytest.warns(
+                UserWarning,
+                match="Attempted to establish mTLS, but a custom async transport was provided",
+            ):
+                await authed_session.configure_mtls_channel(callback)
+
+            assert authed_session._is_mtls is False
+            assert authed_session._cached_cert is None
+
+        await authed_session.close()
+
 
 def test_mock_request_clone():
     request = MockRequest()

packages/google-auth/tests/transport/aio/test_sessions_mtls.py

@@ -193,9 +193,7 @@ async def test_configure_mtls_channel_custom_request(self):
             # If the request handler is not an AiohttpRequest, the library cannot configure
             # the connection to use mTLS, so _is_mtls must be False to reflect this unconfigured state.
             assert session._is_mtls is False
-            mock_make_context.assert_called_once_with(
-                b"fake_cert_data", b"fake_key_data"
-            )
+            mock_make_context.assert_not_called()
             await session.close()
 
     @pytest.mark.asyncio