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

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

@@ -208,7 +208,7 @@ class _MutualTlsAdapter(requests.adapters.HTTPAdapter):
         google.auth.exceptions.MutualTLSChannelError: If the cert or key is invalid.
     """
 
-    def __init__(self, cert, key):
+    def __init__(self, cert, key, **kwargs):
         import certifi
         import ssl
 
@@ -250,7 +250,7 @@ def __init__(self, cert, key):
         self._ctx_poolmanager = ctx_poolmanager
         self._ctx_proxymanager = ctx_proxymanager
 
-        super(_MutualTlsAdapter, self).__init__()
+        super(_MutualTlsAdapter, self).__init__(**kwargs)
 
     def init_poolmanager(self, *args, **kwargs):
         kwargs["ssl_context"] = self._ctx_poolmanager
@@ -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
@@ -475,10 +480,58 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 client_cert_callback
             )
 
+            old_adapter = self.adapters.get("https://")
+
+            kwargs = {}
+            if old_adapter is not None:
+                kwargs["max_retries"] = getattr(old_adapter, "max_retries", 0)
+                kwargs["pool_connections"] = getattr(
+                    old_adapter, "_pool_connections", requests.adapters.DEFAULT_POOLSIZE
+                )
+                kwargs["pool_maxsize"] = getattr(
+                    old_adapter, "_pool_maxsize", requests.adapters.DEFAULT_POOLSIZE
+                )
+                kwargs["pool_block"] = getattr(
+                    old_adapter, "_pool_block", requests.adapters.DEFAULT_POOLBLOCK
+                )
+
+            old_auth_adapter = None
+            auth_kwargs = {}
+            if self._auth_request_session is not None:
+                old_auth_adapter = self._auth_request_session.adapters.get("https://")
+
+                if old_auth_adapter is not None:
+                    auth_kwargs["max_retries"] = getattr(
+                        old_auth_adapter, "max_retries", 0
+                    )
+                    auth_kwargs["pool_connections"] = getattr(
+                        old_auth_adapter,
+                        "_pool_connections",
+                        requests.adapters.DEFAULT_POOLSIZE,
+                    )
+                    auth_kwargs["pool_maxsize"] = getattr(
+                        old_auth_adapter,
+                        "_pool_maxsize",
+                        requests.adapters.DEFAULT_POOLSIZE,
+                    )
+                    auth_kwargs["pool_block"] = getattr(
+                        old_auth_adapter,
+                        "_pool_block",
+                        requests.adapters.DEFAULT_POOLBLOCK,
+                    )
+
             if is_mtls:
-                new_adapter = _MutualTlsAdapter(cert, key)
+                new_adapter = _MutualTlsAdapter(cert, key, **kwargs)
+                if self._auth_request_session is not None:
+                    new_auth_adapter = _MutualTlsAdapter(cert, key, **auth_kwargs)
+                else:
+                    new_auth_adapter = None
             else:
-                new_adapter = requests.adapters.HTTPAdapter()
+                new_adapter = requests.adapters.HTTPAdapter(**kwargs)
+                if self._auth_request_session is not None:
+                    new_auth_adapter = requests.adapters.HTTPAdapter(**auth_kwargs)
+                else:
+                    new_auth_adapter = None
         except (
             exceptions.ClientCertError,
             ImportError,
@@ -489,6 +542,19 @@ def configure_mtls_channel(self, client_cert_callback=None):
             raise new_exc from caught_exc
 
         self.mount("https://", new_adapter)
+
+        if old_adapter is not None and old_adapter is not new_adapter:
+            old_adapter.close()
+
+        if self._auth_request_session is not None and new_auth_adapter is not None:
+            self._auth_request_session.mount("https://", new_auth_adapter)
+
+            if (
+                old_auth_adapter is not None
+                and old_auth_adapter is not new_auth_adapter
+            ):
+                old_auth_adapter.close()
+
         self._is_mtls = is_mtls
         if is_mtls:
             self._cached_cert = cert

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

@@ -335,6 +335,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 `urllib3.PoolManager`.
+            It is not thread-safe to call this explicitly while other
+            threads are making requests.
+
         Returns:
             True if the channel is mutual TLS and False otherwise.
 
@@ -367,9 +372,15 @@ def configure_mtls_channel(self, client_cert_callback=None):
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
+        old_http = self.http
+
         self.http = new_http
         self._is_mtls = new_is_mtls
         self._request.http = new_http
+
+        if old_http is not None and old_http is not new_http:
+            getattr(old_http, "clear", getattr(old_http, "close", lambda: None))()
+
         if new_is_mtls:
             self._cached_cert = cert
         else:
@@ -491,7 +502,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def __del__(self):
         if hasattr(self, "http") and self.http is not None:
-            self.http.clear()
+            getattr(self.http, "clear", getattr(self.http, "close", lambda: None))()
 
     @property
     def headers(self):

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 a separate adapter instance
+        assert isinstance(
+            auth_session._auth_request_session.adapters["https://"],
+            google.auth.transport.requests._MutualTlsAdapter,
+        )
+        assert (
+            auth_session.adapters["https://"]
+            is not 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

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

@@ -243,6 +243,63 @@ def test_configure_mtls_channel_with_metadata(
             cert=pytest.public_cert_bytes, key=pytest.private_key_bytes
         )
 
+    @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True)
+    @mock.patch(
+        "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
+    )
+    def test_configure_mtls_channel_closes_old_poolmanager(
+        self, mock_get_client_cert_and_key, mock_make_mutual_tls_http
+    ):
+        mock_get_client_cert_and_key.return_value = (
+            True,
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+
+        old_http = mock.create_autospec(urllib3.PoolManager)
+        authed_http = google.auth.transport.urllib3.AuthorizedHttp(
+            credentials=mock.Mock(), http=old_http
+        )
+
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            is_mtls = authed_http.configure_mtls_channel()
+
+        assert is_mtls
+        old_http.clear.assert_called_once()
+        mock_make_mutual_tls_http.assert_called_once_with(
+            cert=pytest.public_cert_bytes, key=pytest.private_key_bytes
+        )
+
+    @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True)
+    @mock.patch(
+        "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
+    )
+    def test_configure_mtls_channel_with_none_http(
+        self, mock_get_client_cert_and_key, mock_make_mutual_tls_http
+    ):
+        mock_get_client_cert_and_key.return_value = (
+            True,
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+
+        authed_http = google.auth.transport.urllib3.AuthorizedHttp(
+            credentials=mock.Mock()
+        )
+        authed_http.http = None  # Force old_http to be None
+
+        with mock.patch.dict(
+            os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}
+        ):
+            is_mtls = authed_http.configure_mtls_channel()
+
+        assert is_mtls
+        mock_make_mutual_tls_http.assert_called_once_with(
+            cert=pytest.public_cert_bytes, key=pytest.private_key_bytes
+        )
+
     @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True)
     @mock.patch(
         "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True