fix(transport): propagate mTLS adapter to auth session and fix connection leaks

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

@@ -457,6 +457,11 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 If the callback is None, application default SSL credentials
                 will be used.
 
+        .. warning::
+            Calling this method mutates the underlying `requests.Session` adapter
+            dictionary. It is not thread-safe to call this explicitly while other
+            threads are making requests.
+
         Raises:
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason. The existing session state (such
@@ -488,7 +493,27 @@ def configure_mtls_channel(self, client_cert_callback=None):
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
+        try:
+            old_adapter = self.get_adapter("https://")
+        except requests.exceptions.InvalidSchema:
+            old_adapter = None
+
+        if old_adapter is not None and old_adapter is not new_adapter:
+            old_adapter.close()
+
         self.mount("https://", new_adapter)
+
+        if self._auth_request_session is not None:
+            try:
+                old_auth_adapter = self._auth_request_session.get_adapter("https://")
+            except requests.exceptions.InvalidSchema:
+                old_auth_adapter = None
+
+            if old_auth_adapter is not None and old_auth_adapter is not new_adapter:
+                old_auth_adapter.close()
+
+            self._auth_request_session.mount("https://", new_adapter)
+
         self._is_mtls = is_mtls
         if is_mtls:
             self._cached_cert = cert

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

@@ -453,6 +453,141 @@ def test_configure_mtls_channel_with_metadata(self, mock_get_client_cert_and_key
             google.auth.transport.requests._MutualTlsAdapter,
         )
 
+    @mock.patch(
+        "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
+    )
+    def test_configure_mtls_channel_closes_old_adapters(
+        self, mock_get_client_cert_and_key
+    ):
+        mock_get_client_cert_and_key.return_value = (
+            True,
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+
+        auth_session = google.auth.transport.requests.AuthorizedSession(
+            credentials=mock.Mock()
+        )
+        old_main_adapter = mock.Mock(spec=requests.adapters.HTTPAdapter)
+        old_auth_adapter = mock.Mock(spec=requests.adapters.HTTPAdapter)
+
+        auth_session.mount("https://", old_main_adapter)
+        auth_session._auth_request_session.mount("https://", old_auth_adapter)
+
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            auth_session.configure_mtls_channel()
+
+        old_main_adapter.close.assert_called_once()
+        old_auth_adapter.close.assert_called_once()
+
+    @mock.patch(
+        "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
+    )
+    def test_configure_mtls_channel_mounts_adapter_to_auth_request_session(
+        self, mock_get_client_cert_and_key
+    ):
+        mock_get_client_cert_and_key.return_value = (
+            True,
+            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()
+
+        assert auth_session.is_mtls
+        # Main session gets the mTLS adapter
+        assert isinstance(
+            auth_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+        # _auth_request_session gets the exact same adapter
+        assert isinstance(
+            auth_session._auth_request_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+        assert (
+            auth_session.adapters["https://"]
+            is auth_session._auth_request_session.adapters["https://"]
+        )
+
+    @mock.patch(
+        "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
+    )
+    def test_configure_mtls_channel_without_https_adapter(
+        self, mock_get_client_cert_and_key
+    ):
+        mock_get_client_cert_and_key.return_value = (
+            True,
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+
+        auth_session = google.auth.transport.requests.AuthorizedSession(
+            credentials=mock.Mock()
+        )
+
+        # Remove the 'https://' adapter to trigger InvalidSchema
+        auth_session.adapters.pop("https://", None)
+        auth_session._auth_request_session.adapters.pop("https://", None)
+
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            auth_session.configure_mtls_channel()
+
+        assert auth_session.is_mtls
+        # Main session gets the mTLS adapter
+        assert isinstance(
+            auth_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+        # _auth_request_session gets the exact same adapter
+        assert isinstance(
+            auth_session._auth_request_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+
+    @mock.patch(
+        "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
+    )
+    def test_configure_mtls_channel_without_auth_request_session(
+        self, mock_get_client_cert_and_key
+    ):
+        mock_get_client_cert_and_key.return_value = (
+            True,
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+
+        auth_session = google.auth.transport.requests.AuthorizedSession(
+            credentials=mock.Mock(), auth_request=mock.Mock()
+        )
+        assert auth_session._auth_request_session is None
+
+        old_main_adapter = mock.Mock(spec=requests.adapters.HTTPAdapter)
+        auth_session.mount("https://", old_main_adapter)
+
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            auth_session.configure_mtls_channel()
+
+        old_main_adapter.close.assert_called_once()
+        assert auth_session.is_mtls
+        assert isinstance(
+            auth_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+
     @mock.patch.object(google.auth.transport.requests._MutualTlsAdapter, "__init__")
     @mock.patch(
         "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True