fix(google-auth): fail loudly when dynamically disabling mTLS on active sessions

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

@@ -182,6 +182,13 @@ async def _do_configure():
                     google.auth.transport._mtls_helper.check_use_client_cert
                 )
                 if not use_client_cert:
+                    # Dynamically disabling mTLS on an active session is unsafe in concurrent
+                    # environments and can cause a zombie state mismatch where mTLS contexts
+                    # remain attached while auth checks believe mTLS is disabled.
+                    if getattr(self, "_is_mtls", False):
+                        raise exceptions.MutualTLSChannelError(
+                            "Cannot disable mTLS on an active session. A new AuthorizedSession must be created."
+                        )
                     return
 
                 try:
@@ -191,6 +198,12 @@ async def _do_configure():
                         key,
                     ) = await mtls.get_client_cert_and_key(client_cert_callback)
 
+                    # Prevent mid-lifecycle transition from mTLS-enabled to mTLS-disabled state.
+                    if getattr(self, "_is_mtls", False) and not is_mtls:
+                        raise exceptions.MutualTLSChannelError(
+                            "Cannot disable mTLS on an active session. A new AuthorizedSession must be created."
+                        )
+
                     if is_mtls:
                         # Re-create the auth request with the new SSL context
                         if AIOHTTP_INSTALLED and isinstance(

packages/google-auth/google/auth/transport/requests.py

@@ -469,6 +469,13 @@ def configure_mtls_channel(self, client_cert_callback=None):
         """
         use_client_cert = google.auth.transport._mtls_helper.check_use_client_cert()
         if not use_client_cert:
+            # Dynamically disabling mTLS on an active session is unsafe in concurrent
+            # environments and can cause a zombie state mismatch where mTLS adapters
+            # remain attached while auth checks believe mTLS is disabled.
+            if getattr(self, "_is_mtls", False):
+                raise exceptions.MutualTLSChannelError(
+                    "Cannot disable mTLS on an active session. A new AuthorizedSession must be created."
+                )
             return
 
         try:
@@ -480,6 +487,12 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 client_cert_callback
             )
 
+            # Prevent mid-lifecycle transition from mTLS-enabled to mTLS-disabled state.
+            if getattr(self, "_is_mtls", False) and not is_mtls:
+                raise exceptions.MutualTLSChannelError(
+                    "Cannot disable mTLS on an active session. A new AuthorizedSession must be created."
+                )
+
             old_adapter = self.adapters.get("https://")
 
             kwargs = {}

packages/google-auth/google/auth/transport/urllib3.py

@@ -350,13 +350,26 @@ def configure_mtls_channel(self, client_cert_callback=None):
         """
         use_client_cert = transport._mtls_helper.check_use_client_cert()
         if not use_client_cert:
+            # Dynamically disabling mTLS on an active session is unsafe in concurrent
+            # environments and can cause a zombie state mismatch where mTLS connection
+            # pools remain attached while auth checks believe mTLS is disabled.
+            if getattr(self, "_is_mtls", False):
+                raise exceptions.MutualTLSChannelError(
+                    "Cannot disable mTLS on an active session. A new AuthorizedHttp must be created."
+                )
             return False
 
         try:
             found_cert_key, cert, key = transport._mtls_helper.get_client_cert_and_key(
                 client_cert_callback
             )
 
+            # Prevent mid-lifecycle transition from mTLS-enabled to mTLS-disabled state.
+            if getattr(self, "_is_mtls", False) and not found_cert_key:
+                raise exceptions.MutualTLSChannelError(
+                    "Cannot disable mTLS on an active session. A new AuthorizedHttp must be created."
+                )
+
             if found_cert_key:
                 new_http = _make_mutual_tls_http(cert, key)
                 new_is_mtls = True

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

@@ -344,3 +344,85 @@ async def test_configure_mtls_channel_close_exception_does_not_abort(self):
             assert session._is_mtls is True
             assert session._cached_cert == b"fake_cert_data"
             await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_subsequent_disabled(self):
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch("os.path.exists") as mock_exists, mock.patch(
+            "builtins.open", mock.mock_open(read_data=json.dumps(VALID_WORKLOAD_CONFIG))
+        ), mock.patch(
+            "google.auth.aio.transport.mtls.get_client_cert_and_key"
+        ) as mock_helper, mock.patch(
+            "google.auth.aio.transport.mtls.make_client_cert_ssl_context"
+        ) as mock_make_context, mock.patch(
+            "aiohttp.TCPConnector"
+        ), mock.patch(
+            "aiohttp.ClientSession"
+        ) as mock_session:
+            mock_session.return_value.close = mock.AsyncMock()
+            mock_exists.return_value = True
+            mock_helper.return_value = (True, b"fake_cert_data", b"fake_key_data")
+
+            mock_context = mock.Mock(spec=ssl.SSLContext)
+            mock_make_context.return_value = mock_context
+
+            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_request = session._auth_request
+
+            # Reset task so we trigger a new configuration run
+            session._mtls_init_task = None
+            mock_helper.return_value = (False, None, None)
+
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                await session.configure_mtls_channel()
+
+            assert session._is_mtls is True
+            assert session._auth_request is first_auth_request
+            await session.close()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_subsequent_env_disabled(self):
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch("os.path.exists") as mock_exists, mock.patch(
+            "builtins.open", mock.mock_open(read_data=json.dumps(VALID_WORKLOAD_CONFIG))
+        ), mock.patch(
+            "google.auth.aio.transport.mtls.get_client_cert_and_key"
+        ) as mock_helper, mock.patch(
+            "google.auth.aio.transport.mtls.make_client_cert_ssl_context"
+        ) as mock_make_context, mock.patch(
+            "aiohttp.TCPConnector"
+        ), mock.patch(
+            "aiohttp.ClientSession"
+        ) as mock_session:
+            mock_session.return_value.close = mock.AsyncMock()
+            mock_exists.return_value = True
+            mock_helper.return_value = (True, b"fake_cert_data", b"fake_key_data")
+
+            mock_context = mock.Mock(spec=ssl.SSLContext)
+            mock_make_context.return_value = mock_context
+
+            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_request = session._auth_request
+
+            # Reset task and disable env var
+            session._mtls_init_task = None
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                with mock.patch.dict(
+                    os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}
+                ):
+                    await session.configure_mtls_channel()
+
+            assert session._is_mtls is True
+            assert session._auth_request is first_auth_request
+            await session.close()
+

packages/google-auth/tests/transport/test_requests.py

@@ -1025,22 +1025,54 @@ def test_configure_mtls_channel_subsequent_disabled(self):
 
         assert auth_session.is_mtls
 
-        # 2. Subsequent call returns no client certificate (disabled)
+        # 2. Subsequent call returns no client certificate (disabled) -> raises MutualTLSChannelError
         with mock.patch(
             "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
         ) as mock_get_client_cert_and_key:
             mock_get_client_cert_and_key.return_value = (False, None, None)
 
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                with mock.patch.dict(
+                    os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+                ):
+                    auth_session.configure_mtls_channel()
+
+        # 3. Verify mTLS state and MutualTlsAdapter are preserved
+        assert auth_session.is_mtls
+        assert isinstance(
+            auth_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+
+    def test_configure_mtls_channel_subsequent_env_disabled(self):
+        # 1. Setup successful mTLS configuration
+        mock_callback = mock.Mock()
+        mock_callback.return_value = (
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+        auth_session = google.auth.transport.requests.AuthorizedSession(
+            credentials=mock.Mock()
+        )
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            auth_session.configure_mtls_channel(mock_callback)
+
+        assert auth_session.is_mtls
+
+        # 2. Subsequent call with mTLS disabled via env var -> raises MutualTLSChannelError
+        with pytest.raises(exceptions.MutualTLSChannelError):
             with mock.patch.dict(
-                os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+                os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "false"}
             ):
                 auth_session.configure_mtls_channel()
 
-        # 3. Verify mTLS is disabled and standard HTTPAdapter is restored
-        assert not auth_session.is_mtls
+        # 3. Verify mTLS state and MutualTlsAdapter are preserved
+        assert auth_session.is_mtls
         assert isinstance(
             auth_session.adapters["https://"],
-            requests.adapters.HTTPAdapter,
+            google.auth.transport.requests._MutualTlsAdapter,
         )
 
 

packages/google-auth/tests/transport/test_urllib3.py

@@ -708,18 +708,48 @@ def test_configure_mtls_channel_subsequent_disabled(
         assert is_mtls
         assert authed_http._is_mtls
 
-        # Subsequent call returns no client certificate (disabled)
+        # Subsequent call returns no client certificate -> raises MutualTLSChannelError
         with mock.patch(
             "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
         ) as mock_get_client_cert_and_key:
             mock_get_client_cert_and_key.return_value = (False, None, None)
 
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                with mock.patch.dict(
+                    os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+                ):
+                    authed_http.configure_mtls_channel()
+
+        # Verify mTLS state is preserved
+        assert authed_http._is_mtls
+        assert isinstance(authed_http.http, mock.Mock)
+
+    @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True)
+    def test_configure_mtls_channel_subsequent_env_disabled(
+        self, mock_make_mutual_tls_http
+    ):
+        callback = mock.Mock()
+        callback.return_value = (pytest.public_cert_bytes, pytest.private_key_bytes)
+
+        authed_http = google.auth.transport.urllib3.AuthorizedHttp(
+            credentials=mock.Mock()
+        )
+
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            is_mtls = authed_http.configure_mtls_channel(callback)
+
+        assert is_mtls
+        assert authed_http._is_mtls
+
+        # Subsequent call with mTLS disabled via env var -> raises MutualTLSChannelError
+        with pytest.raises(exceptions.MutualTLSChannelError):
             with mock.patch.dict(
-                os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+                os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "false"}
             ):
-                is_mtls = authed_http.configure_mtls_channel()
+                authed_http.configure_mtls_channel()
 
-        # Verify mTLS is disabled and standard PoolManager is restored
-        assert not is_mtls
-        assert not authed_http._is_mtls
-        assert isinstance(authed_http.http, urllib3.PoolManager)
+        # Verify mTLS state is preserved
+        assert authed_http._is_mtls
+        assert isinstance(authed_http.http, mock.Mock)