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

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

@@ -149,6 +149,19 @@ def __init__(
             )
         self._auth_request = _auth_request
 
+    def _is_mtls_configured(self) -> bool:
+        """Check if mTLS is currently active based on flag, connector SSL context, or cached cert."""
+        if self._is_mtls:
+            return True
+        if self._cached_cert is not None:
+            return True
+        session = getattr(self._auth_request, "session", None)
+        connector = getattr(session, "connector", None)
+        ssl_ctx = getattr(connector, "_ssl", getattr(connector, "ssl", None))
+        if ssl_ctx is not None and not isinstance(ssl_ctx, bool):
+            return True
+        return False
+
     async def configure_mtls_channel(self, client_cert_callback=None):
         """Configure the client certificate and key for SSL connection.
 
@@ -182,6 +195,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 state mismatch where mTLS contexts
+                    # remain attached while auth checks believe mTLS is disabled.
+                    if self._is_mtls_configured():
+                        raise exceptions.MutualTLSChannelError(
+                            "Cannot disable mTLS on an active session. A new AsyncAuthorizedSession must be created."
+                        )
                     return
 
                 try:
@@ -191,6 +211,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 self._is_mtls_configured() and not is_mtls:
+                        raise exceptions.MutualTLSChannelError(
+                            "Cannot disable mTLS on an active session. A new AsyncAuthorizedSession must be created."
+                        )
+
                     if is_mtls:
                         # Re-create the auth request with the new SSL context
                         if AIOHTTP_INSTALLED and isinstance(
@@ -227,6 +253,8 @@ async def _do_configure():
                     else:
                         self._cached_cert = None
 
+                except exceptions.MutualTLSChannelError:
+                    raise
                 except Exception as caught_exc:
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc

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

@@ -441,6 +441,21 @@ def __init__(
                 "https://{}/".format(self._default_host) if self._default_host else None
             )
 
+    def _is_mtls_configured(self) -> bool:
+        """Check if mTLS is currently active based on flag, adapter type, or cached cert."""
+        if self._is_mtls:
+            return True
+        if getattr(self, "_cached_cert", None) is not None:
+            return True
+        adapter = self.adapters.get("https://")
+        if isinstance(adapter, (_MutualTlsAdapter, _MutualTlsOffloadAdapter)):
+            return True
+        if self._auth_request_session is not None:
+            auth_adapter = self._auth_request_session.adapters.get("https://")
+            if isinstance(auth_adapter, (_MutualTlsAdapter, _MutualTlsOffloadAdapter)):
+                return True
+        return False
+
     def configure_mtls_channel(self, client_cert_callback=None):
         """Configure the client certificate and key for SSL connection.
 
@@ -469,6 +484,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 state mismatch where mTLS adapters
+            # remain attached while auth checks believe mTLS is disabled.
+            if self._is_mtls_configured():
+                raise exceptions.MutualTLSChannelError(
+                    "Cannot disable mTLS on an active session. A new AuthorizedSession must be created."
+                )
             return
 
         try:
@@ -480,6 +502,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 self._is_mtls_configured() 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

@@ -319,6 +319,21 @@ def __init__(
 
         super(AuthorizedHttp, self).__init__()
 
+    def _is_mtls_configured(self) -> bool:
+        """Check if mTLS is currently active based on flag, pool ssl_context, or cached cert."""
+        if self._is_mtls:
+            return True
+        if getattr(self, "_cached_cert", None) is not None:
+            return True
+        if (
+            not self._has_user_provided_http
+            and hasattr(self.http, "connection_pool_kw")
+            and isinstance(self.http.connection_pool_kw, dict)
+            and self.http.connection_pool_kw.get("ssl_context") is not None
+        ):
+            return True
+        return False
+
     def configure_mtls_channel(self, client_cert_callback=None):
         """Configures mutual TLS channel using the given client_cert_callback or
         application default SSL credentials.
@@ -350,13 +365,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 state mismatch where mTLS connection
+            # pools remain attached while auth checks believe mTLS is disabled.
+            if self._is_mtls_configured():
+                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 self._is_mtls_configured() 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,101 @@ 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()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_desynchronized_state_raises(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(mock_creds)
+        # Directly set cached cert, leaving _is_mtls False
+        session._cached_cert = b"fake_cert_data"
+        assert not session._is_mtls
+        assert session._is_mtls_configured()
+
+        with pytest.raises(exceptions.MutualTLSChannelError):
+            with mock.patch.dict(
+                os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}
+            ):
+                await session.configure_mtls_channel()
+        await session.close()
+

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

@@ -1025,23 +1025,76 @@ 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,
+        )
+
+    def test_configure_mtls_channel_desynchronized_state_raises(self):
+        auth_session = google.auth.transport.requests.AuthorizedSession(
+            credentials=mock.Mock()
         )
+        # Mount an mTLS adapter manually, leaving _is_mtls False
+        auth_session.mount(
+            "https://",
+            google.auth.transport.requests._MutualTlsAdapter(
+                pytest.public_cert_bytes, pytest.private_key_bytes
+            ),
+        )
+        assert not auth_session.is_mtls
+        assert auth_session._is_mtls_configured()
+
+        # Calling configure_mtls_channel with mTLS disabled in env should raise
+        with pytest.raises(exceptions.MutualTLSChannelError):
+            with mock.patch.dict(
+                os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "false"}
+            ):
+                auth_session.configure_mtls_channel()
 
 
 class TestMutualTlsOffloadAdapter(object):