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

This PR is fixing two issues found in the auth library:

 1. Adapter Connection Leaks
Whenever the `configure_mtls_channel()` method was called, the code
would create a new mTLS `HTTPAdapter` and mount it to `"https://"`.
However, it never closed the *old* adapter it was replacing. Because of
how the underlying `requests` and `urllib3` libraries work, the old
adapter's connection pool was left open, creating dangling sockets and
eventually causing file descriptor exhaustion.
**The Fix:** I updated the method to retrieve the existing `"https://"`
adapter (via `self.get_adapter()`) before replacing it. If an old
adapter is found, we now explicitly call `old_adapter.close()` to shut
down its connection pool and cleanly release the sockets back to the OS.

2. Missing Auth Request Session Update (Issue #17680)
The `AuthorizedSession` class maintains an internal session called
`_auth_request_session`. This hidden session is specifically responsible
for doing background work, like fetching new OAuth tokens or signing IAM
requests. When mTLS was enabled, the new secure adapter was *only*
applied to the main user-facing session, completely missing this
internal session. As a result, critical token refreshes were bypassing
mTLS and failing when they hit strict Certificate-Based Access (CBA)
endpoints.
**The Fix:** I added logic to check if `self._auth_request_session`
exists during the mTLS configuration. If it does, we now apply the exact
same cleanup (closing its old adapter) and then explicitly mount the new
mTLS adapter to it. This guarantees that all background token refreshes
are routed securely over the mTLS channel.

Also added a documentation warning to clarify that dynamically
reconfiguring the channel mutates the underlying session and is not
natively thread-safe.

Author: nbayati

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