From 7877ee8ff485a8db1f8802de2b3da249776d0a6d Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 13:16:23 -0700 Subject: [PATCH 01/19] chore: Add threading lock for mTLS configuration Introduce a lock for thread-safe reauthentication when mTLS parameters change. --- .../google/auth/transport/requests.py | 53 ++++++++++--------- 1 file changed, 28 insertions(+), 25 deletions(-) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 822cf687f5d0..0867575e34c6 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -19,6 +19,7 @@ import functools import http.client as http_client import logging +import threading import numbers import time from typing import Optional @@ -414,6 +415,7 @@ def __init__( self._refresh_timeout = refresh_timeout self._is_mtls = False self._default_host = default_host + self._reauth_lock = threading.Lock() if auth_request is None: self._auth_request_session = requests.Session() @@ -655,33 +657,34 @@ def request( prefix in url for prefix in MTLS_URL_PREFIXES ) if use_mtls: - ( - call_cert_bytes, - call_key_bytes, - cached_fingerprint, - current_cert_fingerprint, - ) = _mtls_helper.check_parameters_for_unauthorized_response( - self._cached_cert - ) - if cached_fingerprint != current_cert_fingerprint: - try: + with self._reauth_lock: + ( + call_cert_bytes, + call_key_bytes, + cached_fingerprint, + current_cert_fingerprint, + ) = _mtls_helper.check_parameters_for_unauthorized_response( + self._cached_cert + ) + if cached_fingerprint != current_cert_fingerprint: + try: + _LOGGER.info( + "Client certificate has changed, reconfiguring mTLS " + "channel." + ) + self.configure_mtls_channel( + lambda: (call_cert_bytes, call_key_bytes) + ) + except Exception as e: + _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) + raise exceptions.MutualTLSChannelError( + "Failed to reconfigure mTLS channel" + ) from e + else: _LOGGER.info( - "Client certificate has changed, reconfiguring mTLS " - "channel." + "Skipping reconfiguration of mTLS channel because the client" + " certificate has not changed." ) - self.configure_mtls_channel( - lambda: (call_cert_bytes, call_key_bytes) - ) - except Exception as e: - _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) - raise exceptions.MutualTLSChannelError( - "Failed to reconfigure mTLS channel" - ) from e - else: - _LOGGER.info( - "Skipping reconfiguration of mTLS channel because the client" - " certificate has not changed." - ) _LOGGER.info( "Refreshing credentials due to a %s response. Attempt %s/%s.", response.status_code, From 5a7ec5c1920dd127a7aab686d2b6c2c14d2e9476 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 13:19:26 -0700 Subject: [PATCH 02/19] chore: For urllib3 Add reauth_lock to manage mTLS reconfiguration Added a reauthentication lock to prevent concurrent reconfiguration of mTLS channel when client certificate changes. --- .../google/auth/transport/urllib3.py | 61 ++++++++++--------- 1 file changed, 32 insertions(+), 29 deletions(-) diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index 18e6128e03bd..3bc35509d200 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -18,6 +18,7 @@ import http.client as http_client import logging +import threading import warnings # Certifi is Mozilla's certificate bundle. Urllib3 needs a certificate bundle @@ -309,6 +310,7 @@ def __init__( # credentials.refresh). self._request = Request(self.http) self._is_mtls = False + self._reauth_lock = threading.Lock() # https://google.aip.dev/auth/4111 # Attempt to use self-signed JWTs when a service account is used. @@ -437,37 +439,38 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): ): if response.status == http_client.UNAUTHORIZED: if use_mtls: - ( - call_cert_bytes, - call_key_bytes, - cached_fingerprint, - current_cert_fingerprint, - ) = _mtls_helper.check_parameters_for_unauthorized_response( - self._cached_cert - ) - if cached_fingerprint != current_cert_fingerprint: - try: - _LOGGER.info( - "Client certificate has changed, reconfiguring mTLS " - "channel." - ) - self.configure_mtls_channel( - client_cert_callback=lambda: ( - call_cert_bytes, - call_key_bytes, + with self._reauth_lock: + ( + call_cert_bytes, + call_key_bytes, + cached_fingerprint, + current_cert_fingerprint, + ) = _mtls_helper.check_parameters_for_unauthorized_response( + self._cached_cert + ) + if cached_fingerprint != current_cert_fingerprint: + try: + _LOGGER.info( + "Client certificate has changed, reconfiguring mTLS " + "channel." ) + self.configure_mtls_channel( + client_cert_callback=lambda: ( + call_cert_bytes, + call_key_bytes, + ) + ) + except Exception as e: + _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) + raise exceptions.MutualTLSChannelError( + "Failed to reconfigure mTLS channel" + ) from e + + else: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because the " + "client certificate has not changed." ) - except Exception as e: - _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) - raise exceptions.MutualTLSChannelError( - "Failed to reconfigure mTLS channel" - ) from e - - else: - _LOGGER.info( - "Skipping reconfiguration of mTLS channel because the " - "client certificate has not changed." - ) _LOGGER.info( "Refreshing credentials due to a %s response. Attempt %s/%s.", From 496a3de1530cffd3ff8fd447dad3b1849f445f42 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 13:34:50 -0700 Subject: [PATCH 03/19] chore: Add test for MTLS reauth lock on unauthorized response Added a test to ensure reauthentication lock is acquired on unauthorized response for MTLS sessions. --- .../tests/transport/test_requests.py | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 2ca1922494ef..0eeaee97a589 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -16,6 +16,7 @@ import functools import http.client as http_client import os +import threading from unittest import mock import freezegun @@ -31,6 +32,7 @@ import google.auth.transport.requests from google.oauth2 import service_account from tests.transport import compliance +import http.client as http_client @pytest.fixture @@ -1109,3 +1111,37 @@ def test_success_should_use_provider( adapter.proxy_manager_for() mock_proxy_manager_for.assert_called_with(ssl_context=adapter._ctx_proxymanager) + +class TestAuthorizedSessionMTLSReauth: + + @mock.patch("google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response") + def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): + credentials = mock.Mock() + session = google.auth.transport.requests.AuthorizedSession(credentials) + + session._is_mtls = True + + mock_response = mock.Mock() + mock_response.status_code = http_client.UNAUTHORIZED + session._auth_request_session.request = mock.Mock(return_value=mock_response) + + mock_lock = mock.MagicMock() + session._reauth_lock = mock_lock + + mock_check_params.return_value = ( + b"new_cert_bytes", + b"new_key_bytes", + "old_fingerprint", + "new_fingerprint", + ) + session.configure_mtls_channel = mock.Mock() + + try: + session.request("GET", "https://example.mtls.googleapis.com/some/endpoint") + except Exception: + pass + + mock_lock.__enter__.assert_called_once() + mock_lock.__exit__.assert_called_once() + mock_check_params.assert_called_once_with(session._cached_cert) + session.configure_mtls_channel.assert_called_once() From 64c11a2c2fcc5350a057c840613c0f192f9a00d1 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 13:36:33 -0700 Subject: [PATCH 04/19] chore: Add test for reauth lock on unauthorized response Added a test to verify reauthentication lock acquisition on unauthorized responses in AuthorizedHttp. --- .../tests/transport/test_urllib3.py | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index e1c92dbebc2c..dc7436fcda7e 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -14,6 +14,7 @@ import http.client as http_client import os +import threading from unittest import mock import pytest # type: ignore @@ -26,6 +27,7 @@ import google.auth.transport.urllib3 from google.oauth2 import service_account from tests.transport import compliance +import http.client as http_client CERT_MOCK_VAL = b"-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n" KEY_MOCK_VAL = b"-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n" @@ -723,3 +725,37 @@ def test_configure_mtls_channel_subsequent_disabled( assert not is_mtls assert not authed_http._is_mtls assert isinstance(authed_http.http, urllib3.PoolManager) + +class TestAuthorizedHttpMTLSReauth: + + @mock.patch("google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response") + def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): + credentials = mock.Mock() + http_obj = urllib3.AuthorizedHttp(credentials) + + http_obj._is_mtls = True + + mock_response = mock.Mock() + mock_response.status = http_client.UNAUTHORIZED + http_obj.http.request = mock.Mock(return_value=mock_response) + + mock_lock = mock.MagicMock() + http_obj._reauth_lock = mock_lock + + mock_check_params.return_value = ( + b"new_cert_bytes", + b"new_key_bytes", + "old_fingerprint", + "new_fingerprint", + ) + http_obj.configure_mtls_channel = mock.Mock() + + try: + http_obj.urlopen("GET", "https://example.mtls.googleapis.com/some/endpoint") + except Exception: + pass + + mock_lock.__enter__.assert_called_once() + mock_lock.__exit__.assert_called_once() + mock_check_params.assert_called_once_with(http_obj._cached_cert) + http_obj.configure_mtls_channel.assert_called_once() From 20bf4517f95ec37ffb656fd5b9679845559a1578 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 15:18:45 -0700 Subject: [PATCH 05/19] fix: Reorder import statements in requests.py --- packages/google-auth/google/auth/transport/requests.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 0867575e34c6..14913568480a 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -19,8 +19,8 @@ import functools import http.client as http_client import logging -import threading import numbers +import threading import time from typing import Optional From 0795590632c00b02dd1e17661a2c0f7f0b39afaf Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 15:20:40 -0700 Subject: [PATCH 06/19] fix: Modify test request URL for MTLS session Updated the request URL in the test case to be more general. --- packages/google-auth/tests/transport/test_requests.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 0eeaee97a589..c528d232b825 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -32,7 +32,6 @@ import google.auth.transport.requests from google.oauth2 import service_account from tests.transport import compliance -import http.client as http_client @pytest.fixture @@ -1137,7 +1136,7 @@ def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): session.configure_mtls_channel = mock.Mock() try: - session.request("GET", "https://example.mtls.googleapis.com/some/endpoint") + session.request("GET", "https://example.mtls.googleapis.com") except Exception: pass From 174d1e9395d307ce98156e3d1021cdfb105a2d0e Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Wed, 19 Aug 2026 15:21:53 -0700 Subject: [PATCH 07/19] fix: Update URL in test for mTLS endpoint for urllib3 --- packages/google-auth/tests/transport/test_urllib3.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index dc7436fcda7e..118dc2101ce8 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -27,7 +27,6 @@ import google.auth.transport.urllib3 from google.oauth2 import service_account from tests.transport import compliance -import http.client as http_client CERT_MOCK_VAL = b"-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n" KEY_MOCK_VAL = b"-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n" @@ -751,7 +750,7 @@ def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): http_obj.configure_mtls_channel = mock.Mock() try: - http_obj.urlopen("GET", "https://example.mtls.googleapis.com/some/endpoint") + http_obj.urlopen("GET", "https://example.mtls.googleapis.com") except Exception: pass From adc682d4800c7d4848869dcb652c8aa5d01a2514 Mon Sep 17 00:00:00 2001 From: Radhika Agrawal Date: Thu, 20 Aug 2026 18:47:30 +0000 Subject: [PATCH 08/19] fix: Fix the lint and unit test scoverage Signed-off-by: Radhika Agrawal --- .../google/auth/transport/requests.py | 4 +- .../google/auth/transport/urllib3.py | 6 ++- .../tests/transport/test_requests.py | 45 ++++++++++--------- .../tests/transport/test_urllib3.py | 39 ++++++++-------- 4 files changed, 48 insertions(+), 46 deletions(-) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 14913568480a..0c7fcc0aecbe 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -676,7 +676,9 @@ def request( lambda: (call_cert_bytes, call_key_bytes) ) except Exception as e: - _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) + _LOGGER.error( + "Failed to reconfigure mTLS channel: %s", e + ) raise exceptions.MutualTLSChannelError( "Failed to reconfigure mTLS channel" ) from e diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index 3bc35509d200..a1627babba33 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -461,11 +461,13 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): ) ) except Exception as e: - _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) + _LOGGER.error( + "Failed to reconfigure mTLS channel: %s", e + ) raise exceptions.MutualTLSChannelError( "Failed to reconfigure mTLS channel" ) from e - + else: _LOGGER.info( "Skipping reconfiguration of mTLS channel because the " diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index c528d232b825..76b08dda4804 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -1111,36 +1111,37 @@ def test_success_should_use_provider( adapter.proxy_manager_for() mock_proxy_manager_for.assert_called_with(ssl_context=adapter._ctx_proxymanager) -class TestAuthorizedSessionMTLSReauth: - @mock.patch("google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response") - def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): +class TestAuthorizedSessionMTLSReauth: + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + @mock.patch("google.auth.transport.requests.requests.Session.request") + def test_reauth_lock_acquired_on_unauthorized( + self, mock_session_request, mock_check_params + ): credentials = mock.Mock() session = google.auth.transport.requests.AuthorizedSession(credentials) - - session._is_mtls = True - + session._is_mtls = True + session._cached_cert = b"cert" mock_response = mock.Mock() mock_response.status_code = http_client.UNAUTHORIZED - session._auth_request_session.request = mock.Mock(return_value=mock_response) - - mock_lock = mock.MagicMock() - session._reauth_lock = mock_lock - + mock_session_request.return_value = mock_response + real_lock = threading.Lock() + session._reauth_lock = real_lock mock_check_params.return_value = ( b"new_cert_bytes", b"new_key_bytes", "old_fingerprint", "new_fingerprint", ) - session.configure_mtls_channel = mock.Mock() - - try: - session.request("GET", "https://example.mtls.googleapis.com") - except Exception: - pass - - mock_lock.__enter__.assert_called_once() - mock_lock.__exit__.assert_called_once() - mock_check_params.assert_called_once_with(session._cached_cert) - session.configure_mtls_channel.assert_called_once() + lock_held_during_call = {"held": False} + + def verify_lock_held(*args, **kwargs): + lock_held_during_call["held"] = real_lock.locked() + + session.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held) + session.request("GET", "https://example.mtls.googleapis.com/") + + session.configure_mtls_channel.assert_called() + assert lock_held_during_call["held"] is True diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index 118dc2101ce8..ebd7a322e3ad 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -725,36 +725,33 @@ def test_configure_mtls_channel_subsequent_disabled( assert not authed_http._is_mtls assert isinstance(authed_http.http, urllib3.PoolManager) -class TestAuthorizedHttpMTLSReauth: - @mock.patch("google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response") +class TestAuthorizedHttpMTLSReauth: + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): credentials = mock.Mock() - http_obj = urllib3.AuthorizedHttp(credentials) - + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) http_obj._is_mtls = True - + http_obj._cached_cert = b"cert" mock_response = mock.Mock() mock_response.status = http_client.UNAUTHORIZED - http_obj.http.request = mock.Mock(return_value=mock_response) - - mock_lock = mock.MagicMock() - http_obj._reauth_lock = mock_lock - + http_obj.http.urlopen = mock.Mock(return_value=mock_response) + real_lock = threading.Lock() + http_obj._reauth_lock = real_lock mock_check_params.return_value = ( b"new_cert_bytes", b"new_key_bytes", "old_fingerprint", "new_fingerprint", ) - http_obj.configure_mtls_channel = mock.Mock() - - try: - http_obj.urlopen("GET", "https://example.mtls.googleapis.com") - except Exception: - pass - - mock_lock.__enter__.assert_called_once() - mock_lock.__exit__.assert_called_once() - mock_check_params.assert_called_once_with(http_obj._cached_cert) - http_obj.configure_mtls_channel.assert_called_once() + lock_held_during_call = {"held": False} + + def verify_lock_held(*args, **kwargs): + lock_held_during_call["held"] = real_lock.locked() + + http_obj.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held) + http_obj.request("GET", "https://example.mtls.googleapis.com/") + http_obj.configure_mtls_channel.assert_called() + assert lock_held_during_call["held"] is True From 1035172d33562eb6cdddf5ade7012bae3557e681 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Mon, 31 Aug 2026 11:43:13 -0700 Subject: [PATCH 09/19] chore: Implement MTLS URL prefix handling Add support for MTLS URL prefixes in authentication --- packages/google-auth/google/auth/transport/urllib3.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index a1627babba33..3ca2fe94ab58 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -438,6 +438,11 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): and _credential_refresh_attempt < self._max_refresh_attempts ): if response.status == http_client.UNAUTHORIZED: + MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"] + use_mtls = getattr(self, "_is_mtls", False) and any( + prefix in url for prefix in MTLS_URL_PREFIXES + ) + if use_mtls: with self._reauth_lock: ( From fcdba65fa43d4fefabe780903b86de3aeb7a6946 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Mon, 31 Aug 2026 11:45:17 -0700 Subject: [PATCH 10/19] chore: Refactor test for MTLS session reauthentication Refactor test to use mock response directly and verify lock behavior. --- .../tests/transport/test_requests.py | 37 ++++++++++++++++--- 1 file changed, 32 insertions(+), 5 deletions(-) diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 76b08dda4804..b8b692bcdc12 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -1124,10 +1124,9 @@ def test_reauth_lock_acquired_on_unauthorized( session = google.auth.transport.requests.AuthorizedSession(credentials) session._is_mtls = True session._cached_cert = b"cert" - mock_response = mock.Mock() - mock_response.status_code = http_client.UNAUTHORIZED - mock_session_request.return_value = mock_response - real_lock = threading.Lock() + mock_response = mock.Mock(status_code=http_client.UNAUTHORIZED) + mock_success_response = mock.Mock(status_code=http_client.OK) + mock_session_request.side_effect = [mock_response, mock_success_response] session._reauth_lock = real_lock mock_check_params.return_value = ( b"new_cert_bytes", @@ -1138,10 +1137,38 @@ def test_reauth_lock_acquired_on_unauthorized( lock_held_during_call = {"held": False} def verify_lock_held(*args, **kwargs): - lock_held_during_call["held"] = real_lock.locked() + lock_held_during_call["held"] = session._reauth_lock.locked() session.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held) session.request("GET", "https://example.mtls.googleapis.com/") session.configure_mtls_channel.assert_called() assert lock_held_during_call["held"] is True + + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + @mock.patch("google.auth.transport.requests.requests.Session.request") + def test_reauth_skipped_when_cert_fingerprint_matches( + self, mock_session_request, mock_check_params + ): + credentials = mock.Mock() + session = google.auth.transport.requests.AuthorizedSession(credentials) + session._is_mtls = True + session._cached_cert = b"cert" + + mock_session_request.side_effect = [ + mock.Mock(status_code=http_client.UNAUTHORIZED), + mock.Mock(status_code=http_client.OK), + ] + mock_check_params.return_value = ( + b"same_cert_bytes", + b"same_key_bytes", + "same_fingerprint", + "same_fingerprint", + ) + session.configure_mtls_channel = mock.Mock() + + session.request("GET", "https://example.mtls.googleapis.com/") + + session.configure_mtls_channel.assert_not_called() From dda49ebe041ab5f88386abf5baf220e96c5931ec Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Mon, 31 Aug 2026 13:45:20 -0700 Subject: [PATCH 11/19] fix: Add reauth_lock to session in test_requests.py --- packages/google-auth/tests/transport/test_requests.py | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index b8b692bcdc12..1ac44db55cd4 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -1127,6 +1127,7 @@ def test_reauth_lock_acquired_on_unauthorized( mock_response = mock.Mock(status_code=http_client.UNAUTHORIZED) mock_success_response = mock.Mock(status_code=http_client.OK) mock_session_request.side_effect = [mock_response, mock_success_response] + real_lock = threading.Lock() session._reauth_lock = real_lock mock_check_params.return_value = ( b"new_cert_bytes", From 8376f337491f3aaf2e8ba16b45219d93ef545925 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Mon, 31 Aug 2026 14:11:06 -0700 Subject: [PATCH 12/19] fix: fix spacing for skipping lint error Log message added to indicate skipping mTLS reconfiguration due to unchanged client certificate. --- packages/google-auth/google/auth/transport/urllib3.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index 3ca2fe94ab58..c9d442ecff8a 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -472,7 +472,6 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): raise exceptions.MutualTLSChannelError( "Failed to reconfigure mTLS channel" ) from e - else: _LOGGER.info( "Skipping reconfiguration of mTLS channel because the " From aa4c53551f2a14b480a0cd95ad02a6d85bfd9506 Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Mon, 31 Aug 2026 14:12:29 -0700 Subject: [PATCH 13/19] fix: Clean up whitespace in test_requests.py Removed unnecessary blank lines in test_requests.py. --- packages/google-auth/tests/transport/test_requests.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 1ac44db55cd4..85c86e29111b 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -1157,7 +1157,7 @@ def test_reauth_skipped_when_cert_fingerprint_matches( session = google.auth.transport.requests.AuthorizedSession(credentials) session._is_mtls = True session._cached_cert = b"cert" - + mock_session_request.side_effect = [ mock.Mock(status_code=http_client.UNAUTHORIZED), mock.Mock(status_code=http_client.OK), @@ -1169,7 +1169,7 @@ def test_reauth_skipped_when_cert_fingerprint_matches( "same_fingerprint", ) session.configure_mtls_channel = mock.Mock() - + session.request("GET", "https://example.mtls.googleapis.com/") session.configure_mtls_channel.assert_not_called() From 358177295e8fed6cd64eb72eb5e978fb56694c0e Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Mon, 31 Aug 2026 14:24:16 -0700 Subject: [PATCH 14/19] fix: Clean up whitespace in urllib3.py Remove unnecessary blank line in urllib3.py --- packages/google-auth/google/auth/transport/urllib3.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index c9d442ecff8a..9f8add47cdae 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -442,7 +442,7 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): use_mtls = getattr(self, "_is_mtls", False) and any( prefix in url for prefix in MTLS_URL_PREFIXES ) - + if use_mtls: with self._reauth_lock: ( From 85a5dab421ccc8a22df06b1aedbb68da58a6b59a Mon Sep 17 00:00:00 2001 From: Radhika Agrawal Date: Mon, 31 Aug 2026 21:36:34 +0000 Subject: [PATCH 15/19] fix: fix all the lint errors via blacken Signed-off-by: Radhika Agrawal --- packages/google-auth/google/auth/transport/urllib3.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index 9f8add47cdae..e6ce8d8b359b 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -438,7 +438,10 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): and _credential_refresh_attempt < self._max_refresh_attempts ): if response.status == http_client.UNAUTHORIZED: - MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"] + MTLS_URL_PREFIXES = [ + "mtls.googleapis.com", + "mtls.sandbox.googleapis.com", + ] use_mtls = getattr(self, "_is_mtls", False) and any( prefix in url for prefix in MTLS_URL_PREFIXES ) From a273fba1f5da357f8326592ace723c69cceef03c Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Tue, 1 Sep 2026 21:11:05 -0700 Subject: [PATCH 16/19] fix: Implement test for skipped reauth with matching cert Add test for reauthentication when certificate fingerprint matches. --- .../tests/transport/test_urllib3.py | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index ebd7a322e3ad..c6ef0d193401 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -755,3 +755,34 @@ def verify_lock_held(*args, **kwargs): http_obj.request("GET", "https://example.mtls.googleapis.com/") http_obj.configure_mtls_channel.assert_called() assert lock_held_during_call["held"] is True + + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + def test_reauth_skipped_when_cert_fingerprint_matches(self, mock_check_params): + credentials = mock.Mock() + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) + http_obj._is_mtls = True + http_obj._cached_cert = b"cert" + + mock_response_unauth = mock.Mock() + mock_response_unauth.status = http_client.UNAUTHORIZED + mock_response_ok = mock.Mock() + mock_response_ok.status = http_client.OK + + http_obj.http.urlopen = mock.Mock( + side_effect=[mock_response_unauth, mock_response_ok] + ) + + mock_check_params.return_value = ( + b"same_cert_bytes", + b"same_key_bytes", + "same_fingerprint", + "same_fingerprint", + ) + http_obj.configure_mtls_channel = mock.Mock() + + http_obj.request("GET", "https://example.mtls.googleapis.com/") + + http_obj.configure_mtls_channel.assert_not_called() + From bcef1ea4ae900ddd03b0c4b0d87fe760c9b9ff6f Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Tue, 1 Sep 2026 21:48:11 -0700 Subject: [PATCH 17/19] fix: Refactor test_urllib3.py by removing unnecessary lines fix: Refactor test_urllib3.py by removing unnecessary lines --- packages/google-auth/tests/transport/test_urllib3.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index c6ef0d193401..a0af90bc8c95 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -764,16 +764,13 @@ def test_reauth_skipped_when_cert_fingerprint_matches(self, mock_check_params): http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) http_obj._is_mtls = True http_obj._cached_cert = b"cert" - mock_response_unauth = mock.Mock() mock_response_unauth.status = http_client.UNAUTHORIZED mock_response_ok = mock.Mock() - mock_response_ok.status = http_client.OK - + mock_response_ok.status = http_client.OK http_obj.http.urlopen = mock.Mock( side_effect=[mock_response_unauth, mock_response_ok] ) - mock_check_params.return_value = ( b"same_cert_bytes", b"same_key_bytes", @@ -781,8 +778,5 @@ def test_reauth_skipped_when_cert_fingerprint_matches(self, mock_check_params): "same_fingerprint", ) http_obj.configure_mtls_channel = mock.Mock() - http_obj.request("GET", "https://example.mtls.googleapis.com/") - http_obj.configure_mtls_channel.assert_not_called() - From 3489c0626dbe054edc863964591826cd16fb9a6d Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Tue, 1 Sep 2026 21:55:49 -0700 Subject: [PATCH 18/19] fix: Remove obsolete tests from test_urllib3.py fix: Removed several tests related to timeout, certifi, and credential stubs from test_urllib3.py. --- .../tests/transport/test_urllib3.py | 744 +----------------- 1 file changed, 1 insertion(+), 743 deletions(-) diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index a0af90bc8c95..2f9e5b1398fc 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -37,746 +37,4 @@ def make_request(self): http = urllib3.PoolManager() return google.auth.transport.urllib3.Request(http) - def test_timeout(self): - http = mock.create_autospec(urllib3.PoolManager) - request = google.auth.transport.urllib3.Request(http) - request(url="http://example.com", method="GET", timeout=5) - - assert http.request.call_args[1]["timeout"] == 5 - - -def test__make_default_http_with_certifi(): - http = google.auth.transport.urllib3._make_default_http() - assert "cert_reqs" in http.connection_pool_kw - - -@mock.patch.object(google.auth.transport.urllib3, "certifi", new=None) -def test__make_default_http_without_certifi(): - http = google.auth.transport.urllib3._make_default_http() - assert "cert_reqs" not in http.connection_pool_kw - - -class CredentialsStub(google.auth.credentials.Credentials): - def __init__(self, token="token"): - super(CredentialsStub, self).__init__() - self.token = token - - def apply(self, headers, token=None): - headers["authorization"] = self.token - - def before_request(self, request, method, url, headers): - self.apply(headers) - - def refresh(self, request): - self.token += "1" - - def with_quota_project(self, quota_project_id): - raise NotImplementedError() - - -class HttpStub(object): - def __init__(self, responses, headers=None): - self.responses = responses - self.requests = [] - self.headers = headers or {} - - def urlopen(self, method, url, body=None, headers=None, **kwargs): - self.requests.append((method, url, body, headers, kwargs)) - return self.responses.pop(0) - - def clear(self): - pass - - -class ResponseStub(object): - def __init__(self, status=http_client.OK, data=None): - self.status = status - self.data = data - - -class TestMakeMutualTlsHttp(object): - def test_success(self): - http = google.auth.transport.urllib3._make_mutual_tls_http( - pytest.public_cert_bytes, pytest.private_key_bytes - ) - assert isinstance(http, urllib3.PoolManager) - - def test_crypto_error(self): - with pytest.raises(exceptions.MutualTLSChannelError): - google.auth.transport.urllib3._make_mutual_tls_http( - b"invalid cert", b"invalid key" - ) - - @mock.patch("google.auth.transport.urllib3._mtls_helper.secure_cert_key_paths") - def test_setup_error_raises_mutual_tls_channel_error(self, mock_secure_paths): - mock_secure_paths.side_effect = OSError("Disk full") - with pytest.raises(exceptions.MutualTLSChannelError) as exc_info: - google.auth.transport.urllib3._make_mutual_tls_http(b"cert", b"key") - assert "Failed to configure client certificate" in str(exc_info.value) - assert isinstance(exc_info.value.__cause__, OSError) - - -class TestAuthorizedHttp(object): - TEST_URL = "http://example.com" - - def test_authed_http_defaults(self): - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - mock.sentinel.credentials - ) - - assert authed_http.credentials == mock.sentinel.credentials - assert isinstance(authed_http.http, urllib3.PoolManager) - - def test_urlopen_no_refresh(self): - credentials = mock.Mock(wraps=CredentialsStub()) - response = ResponseStub() - http = HttpStub([response]) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - - result = authed_http.urlopen("GET", self.TEST_URL) - - assert result == response - assert credentials.before_request.called - assert not credentials.refresh.called - assert http.requests == [ - ("GET", self.TEST_URL, None, {"authorization": "token"}, {}) - ] - - def test_urlopen_refresh(self): - credentials = mock.Mock(wraps=CredentialsStub()) - final_response = ResponseStub(status=http_client.OK) - # First request will 401, second request will succeed. - http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED), final_response]) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - - authed_http = authed_http.urlopen("GET", "http://example.com") - - assert authed_http == final_response - assert credentials.before_request.call_count == 2 - assert credentials.refresh.called - assert http.requests == [ - ("GET", self.TEST_URL, None, {"authorization": "token"}, {}), - ("GET", self.TEST_URL, None, {"authorization": "token1"}, {}), - ] - - def test_urlopen_no_default_host(self): - credentials = mock.create_autospec(service_account.Credentials) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials) - - authed_http.credentials._create_self_signed_jwt.assert_called_once_with(None) - - def test_urlopen_with_default_host(self): - default_host = "pubsub.googleapis.com" - credentials = mock.create_autospec(service_account.Credentials) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, default_host=default_host - ) - - authed_http.credentials._create_self_signed_jwt.assert_called_once_with( - "https://{}/".format(default_host) - ) - - def test_proxies(self): - http = mock.create_autospec(urllib3.PoolManager) - authed_http = google.auth.transport.urllib3.AuthorizedHttp(None, http=http) - - with authed_http: - pass - - assert http.__enter__.called - assert http.__exit__.called - - authed_http.headers = mock.sentinel.headers - assert authed_http.headers == http.headers - - @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) - def test_configure_mtls_channel_with_callback(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(), http=mock.Mock() - ) - - with pytest.warns(UserWarning): - 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 - 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_metadata( - self, mock_get_client_cert_and_key, mock_make_mutual_tls_http - ): - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials=mock.Mock() - ) - - mock_get_client_cert_and_key.return_value = ( - True, - pytest.public_cert_bytes, - pytest.private_key_bytes, - ) - 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_get_client_cert_and_key.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_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 - ) - def test_configure_mtls_channel_non_mtls( - self, mock_get_client_cert_and_key, mock_make_mutual_tls_http - ): - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials=mock.Mock() - ) - - mock_get_client_cert_and_key.return_value = (False, None, None) - with mock.patch.dict( - os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} - ): - is_mtls = authed_http.configure_mtls_channel() - - assert not is_mtls - # If client certificate and key are not found, the transport falls back to - # a standard connection. _is_mtls must be False to reflect this fallback state. - assert authed_http._is_mtls is False - mock_get_client_cert_and_key.assert_called_once() - mock_make_mutual_tls_http.assert_not_called() - - @mock.patch( - "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True - ) - def test_configure_mtls_channel_exceptions(self, mock_get_client_cert_and_key): - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials=mock.Mock() - ) - - mock_get_client_cert_and_key.side_effect = exceptions.ClientCertError() - with pytest.raises(exceptions.MutualTLSChannelError): - with mock.patch.dict( - os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} - ): - authed_http.configure_mtls_channel() - assert authed_http._is_mtls is False - - mock_get_client_cert_and_key.side_effect = OSError("Mock file read error") - with pytest.raises(exceptions.MutualTLSChannelError): - with mock.patch.dict( - os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} - ): - authed_http.configure_mtls_channel() - assert authed_http._is_mtls is False - - @mock.patch( - "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True - ) - @mock.patch( - "google.auth.transport.urllib3.urllib3.util.ssl_.create_urllib3_context", - autospec=True, - ) - def test_configure_mtls_channel_cert_loading_exceptions( - self, mock_create_urllib3_context, mock_get_client_cert_and_key - ): - import ssl - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials=mock.Mock() - ) - - mock_get_client_cert_and_key.return_value = ( - True, - pytest.public_cert_bytes, - pytest.private_key_bytes, - ) - - for exception_type in [ValueError("error"), ssl.SSLError("error")]: - mock_ctx = mock.Mock() - mock_ctx.load_cert_chain.side_effect = exception_type - mock_create_urllib3_context.return_value = mock_ctx - - with pytest.raises(exceptions.MutualTLSChannelError): - with mock.patch.dict( - os.environ, - {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}, - ): - authed_http.configure_mtls_channel() - assert authed_http._is_mtls is False - - assert not authed_http._is_mtls - - @mock.patch( - "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True - ) - @mock.patch.dict( - os.environ, - { - "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false", - "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false", - "GOOGLE_API_CERTIFICATE_CONFIG": "", - "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", - }, - ) - def test_configure_mtls_channel_without_client_cert_env( - self, get_client_cert_and_key - ): - callback = mock.Mock() - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials=mock.Mock(), http=mock.Mock() - ) - - env_to_patch = { - environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "", - environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE: "", - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG: "", - environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH: "", - } - with mock.patch.dict(os.environ, env_to_patch): - # Test the callback is not called if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. - is_mtls = authed_http.configure_mtls_channel(callback) - assert not is_mtls - callback.assert_not_called() - - # Test ADC client cert is not used if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. - is_mtls = authed_http.configure_mtls_channel(callback) - assert not is_mtls - get_client_cert_and_key.assert_not_called() - - def test_clear_pool_on_del(self): - http = mock.create_autospec(urllib3.PoolManager) - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - mock.sentinel.credentials, http=http - ) - authed_http.__del__() - http.clear.assert_called_with() - - authed_http.http = None - authed_http.__del__() - # Expect it to not crash - - @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) - @mock.patch("google.auth.transport.urllib3._make_default_http", autospec=True) - def test_cert_rotation_when_cert_mismatch_and_mtls_endpoint_used( - self, mock_make_default_http, mock_make_mutual_tls_http - ): - credentials = mock.Mock(wraps=CredentialsStub()) - final_response = ResponseStub(status=http_client.OK) - - # We simulate the HTTP stub rotation. When mTLS http is created, we return rotated_http. - rotated_http = HttpStub([final_response]) - mock_make_mutual_tls_http.return_value = rotated_http - - http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - - old_cert = b"-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n" - - # New certificate and key to simulate rotation. - new_cert = pytest.public_cert_bytes - new_key = pytest.private_key_bytes - # Set _cached_cert to a callable that returns the old certificate. - authed_http._cached_cert = old_cert - authed_http._is_mtls = True - # Mock call_client_cert_callback to return the new certificate. - with mock.patch.object( - google.auth.transport._mtls_helper, - "call_client_cert_callback", - return_value=(new_cert, new_key), - ) as mock_callback: - # mTLS endpoint is used, and client cert env var is true - with mock.patch.dict( - os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} - ): - result = authed_http.urlopen( - "GET", "http://example.mtls.googleapis.com" - ) - - # Asserts to verify the behavior. - assert result == final_response - assert credentials.refresh.called - assert credentials.refresh.call_count == 1 - assert mock_callback.called - mock_make_mutual_tls_http.assert_called_once_with(cert=new_cert, key=new_key) - - def test_no_cert_rotation_when_cert_match_and_mtls_endpoint_used(self): - credentials = mock.Mock(wraps=CredentialsStub()) - final_response = ResponseStub(status=http_client.UNAUTHORIZED) - http = HttpStub( - [ - ResponseStub(status=http_client.UNAUTHORIZED), - ResponseStub(status=http_client.UNAUTHORIZED), - ResponseStub(status=http_client.UNAUTHORIZED), - ] - ) - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - old_cert = CERT_MOCK_VAL - - new_cert = old_cert - new_key = KEY_MOCK_VAL - # Set _cached_cert to a callable that returns the same certificate. - authed_http._cached_cert = old_cert - authed_http._is_mtls = True - # Mock call_client_cert_callback to return the certificate. - with mock.patch.object( - google.auth.transport._mtls_helper, - "call_client_cert_callback", - return_value=(new_cert, new_key), - ): - # mTLS endpoint is used - result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") - - # Asserts to verify the behavior. - assert credentials.refresh.call_count == 2 - assert result.status == final_response.status - - def test_no_cert_match_check_when_mtls_endpoint_not_used(self): - credentials = mock.Mock(wraps=CredentialsStub()) - final_response = ResponseStub(status=http_client.UNAUTHORIZED) - http = HttpStub( - [ - ResponseStub(status=http_client.UNAUTHORIZED), - ResponseStub(status=http_client.UNAUTHORIZED), - ResponseStub(status=http_client.UNAUTHORIZED), - ] - ) - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - authed_http._is_mtls = False - new_cert = CERT_MOCK_VAL - new_key = KEY_MOCK_VAL - - # Mock call_client_cert_callback to return the certificate. - with mock.patch.object( - google.auth.transport._mtls_helper, - "call_client_cert_callback", - return_value=(new_cert, new_key), - ) as mock_callback: - # non-mTLS endpoint is used - result = authed_http.urlopen("GET", "http://example.googleapis.com") - - # Asserts to verify the behavior. - assert not mock_callback.called - assert result.status == final_response.status - - def test_no_cert_rotation_when_no_unauthorized_response(self): - credentials = mock.Mock(wraps=CredentialsStub()) - final_response = ResponseStub(status=http_client.UPGRADE_REQUIRED) - - # Response is set to code other than 401(Unauthorized). - http = HttpStub([ResponseStub(status=http_client.UPGRADE_REQUIRED)]) - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - authed_http._is_mtls = True - with mock.patch.dict( - os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} - ): - # mTLS endpoint is used - result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") - assert result.status == final_response.status - assert not credentials.refresh.called - assert credentials.refresh.call_count == 0 - - def test_cert_rotation_failure_raises_error(self): - credentials = mock.Mock(wraps=CredentialsStub()) - http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - - old_cert = b"-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n" - - # New certificate and key to simulate rotation. - new_cert = CERT_MOCK_VAL - new_key = KEY_MOCK_VAL - authed_http._cached_cert = old_cert - authed_http._is_mtls = True - - # Mock call_client_cert_callback to return the new certificate. - with mock.patch.object( - google.auth.transport._mtls_helper, - "check_parameters_for_unauthorized_response", - return_value=(new_cert, new_key, "old_fingerprint", "new_fingerprint"), - ) as mock_check_params: - with mock.patch.object( - authed_http, - "configure_mtls_channel", - side_effect=Exception("Failed to reconfigure"), - ) as mock_reconfigure: - with pytest.raises(exceptions.MutualTLSChannelError): - authed_http.urlopen("GET", "https://example.mtls.googleapis.com") - - mock_check_params.assert_called_once() - mock_reconfigure.assert_called_once() - credentials.refresh.assert_not_called() - - def test_cert_rotation_check_params_fails(self): - credentials = mock.Mock(wraps=CredentialsStub()) - http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http - ) - authed_http._is_mtls = True - authed_http._cached_cert = b"cached_cert" - - with mock.patch( - "google.auth.transport.urllib3._mtls_helper.check_parameters_for_unauthorized_response", - side_effect=Exception("check_params failed"), - ) as mock_check_params: - with pytest.raises(Exception, match="check_params failed"): - authed_http.urlopen("GET", "http://example.mtls.googleapis.com") - - mock_check_params.assert_called_once() - credentials.refresh.assert_not_called() - - def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): - """ - Tests that the code can handle a refresh triggered by a status code - other than 401 (UNAUTHORIZED). This covers the 'else' branch of the - 'if response.status_code == http_client.UNAUTHORIZED' check - """ - credentials = mock.Mock(wraps=CredentialsStub()) - # Configure the session to treat 503 (Service Unavailable) as a refreshable error - custom_codes = [http_client.SERVICE_UNAVAILABLE] - - # Return 503 first, then 200 - http = HttpStub( - [ - ResponseStub(status=http_client.SERVICE_UNAVAILABLE), - ResponseStub(status=http_client.OK), - ] - ) - - authed_http = google.auth.transport.urllib3.AuthorizedHttp( - credentials, http=http, refresh_status_codes=custom_codes - ) - - # Enable mTLS to prove it is skipped despite being enabled - authed_http._is_mtls = True - mtls_url = "https://mtls.googleapis.com/test" - - with mock.patch( - "google.auth.transport.urllib3._mtls_helper", autospec=True - ) as mock_helper: - authed_http.urlopen("GET", mtls_url) - - # Assert refresh happened (Outer Check was True) - assert credentials.refresh.called - - # Assert mTLS check logic was SKIPPED (Inner Check was False) - assert not mock_helper.check_parameters_for_unauthorized_response.called - - @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) - def test_configure_mtls_channel_subsequent_failure(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 fails - 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.side_effect = exceptions.ClientCertError() - - 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 it retains its previous mTLS state and connection pool - 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_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 returns no client certificate (disabled) - 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 mock.patch.dict( - os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} - ): - is_mtls = 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) - - -class TestAuthorizedHttpMTLSReauth: - @mock.patch( - "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" - ) - def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): - credentials = mock.Mock() - http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) - http_obj._is_mtls = True - http_obj._cached_cert = b"cert" - mock_response = mock.Mock() - mock_response.status = http_client.UNAUTHORIZED - http_obj.http.urlopen = mock.Mock(return_value=mock_response) - real_lock = threading.Lock() - http_obj._reauth_lock = real_lock - mock_check_params.return_value = ( - b"new_cert_bytes", - b"new_key_bytes", - "old_fingerprint", - "new_fingerprint", - ) - lock_held_during_call = {"held": False} - - def verify_lock_held(*args, **kwargs): - lock_held_during_call["held"] = real_lock.locked() - - http_obj.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held) - http_obj.request("GET", "https://example.mtls.googleapis.com/") - http_obj.configure_mtls_channel.assert_called() - assert lock_held_during_call["held"] is True - - @mock.patch( - "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" - ) - def test_reauth_skipped_when_cert_fingerprint_matches(self, mock_check_params): - credentials = mock.Mock() - http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) - http_obj._is_mtls = True - http_obj._cached_cert = b"cert" - mock_response_unauth = mock.Mock() - mock_response_unauth.status = http_client.UNAUTHORIZED - mock_response_ok = mock.Mock() - mock_response_ok.status = http_client.OK - http_obj.http.urlopen = mock.Mock( - side_effect=[mock_response_unauth, mock_response_ok] - ) - mock_check_params.return_value = ( - b"same_cert_bytes", - b"same_key_bytes", - "same_fingerprint", - "same_fingerprint", - ) - http_obj.configure_mtls_channel = mock.Mock() - http_obj.request("GET", "https://example.mtls.googleapis.com/") - http_obj.configure_mtls_channel.assert_not_called() + def test From be7908b624572cdaf4599ee1c7b2b677d82e277d Mon Sep 17 00:00:00 2001 From: agrawalradhika-cell Date: Tue, 1 Sep 2026 21:56:48 -0700 Subject: [PATCH 19/19] fix: Implement timeout test for urllib3 Request Add test for timeout in urllib3 Request --- .../tests/transport/test_urllib3.py | 744 +++++++++++++++++- 1 file changed, 743 insertions(+), 1 deletion(-) diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index 2f9e5b1398fc..e4a0fe20322b 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -37,4 +37,746 @@ def make_request(self): http = urllib3.PoolManager() return google.auth.transport.urllib3.Request(http) - def test + def test_timeout(self): + http = mock.create_autospec(urllib3.PoolManager) + request = google.auth.transport.urllib3.Request(http) + request(url="http://example.com", method="GET", timeout=5) + + assert http.request.call_args[1]["timeout"] == 5 + + +def test__make_default_http_with_certifi(): + http = google.auth.transport.urllib3._make_default_http() + assert "cert_reqs" in http.connection_pool_kw + + +@mock.patch.object(google.auth.transport.urllib3, "certifi", new=None) +def test__make_default_http_without_certifi(): + http = google.auth.transport.urllib3._make_default_http() + assert "cert_reqs" not in http.connection_pool_kw + + +class CredentialsStub(google.auth.credentials.Credentials): + def __init__(self, token="token"): + super(CredentialsStub, self).__init__() + self.token = token + + def apply(self, headers, token=None): + headers["authorization"] = self.token + + def before_request(self, request, method, url, headers): + self.apply(headers) + + def refresh(self, request): + self.token += "1" + + def with_quota_project(self, quota_project_id): + raise NotImplementedError() + + +class HttpStub(object): + def __init__(self, responses, headers=None): + self.responses = responses + self.requests = [] + self.headers = headers or {} + + def urlopen(self, method, url, body=None, headers=None, **kwargs): + self.requests.append((method, url, body, headers, kwargs)) + return self.responses.pop(0) + + def clear(self): + pass + + +class ResponseStub(object): + def __init__(self, status=http_client.OK, data=None): + self.status = status + self.data = data + + +class TestMakeMutualTlsHttp(object): + def test_success(self): + http = google.auth.transport.urllib3._make_mutual_tls_http( + pytest.public_cert_bytes, pytest.private_key_bytes + ) + assert isinstance(http, urllib3.PoolManager) + + def test_crypto_error(self): + with pytest.raises(exceptions.MutualTLSChannelError): + google.auth.transport.urllib3._make_mutual_tls_http( + b"invalid cert", b"invalid key" + ) + + @mock.patch("google.auth.transport.urllib3._mtls_helper.secure_cert_key_paths") + def test_setup_error_raises_mutual_tls_channel_error(self, mock_secure_paths): + mock_secure_paths.side_effect = OSError("Disk full") + with pytest.raises(exceptions.MutualTLSChannelError) as exc_info: + google.auth.transport.urllib3._make_mutual_tls_http(b"cert", b"key") + assert "Failed to configure client certificate" in str(exc_info.value) + assert isinstance(exc_info.value.__cause__, OSError) + + +class TestAuthorizedHttp(object): + TEST_URL = "http://example.com" + + def test_authed_http_defaults(self): + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + mock.sentinel.credentials + ) + + assert authed_http.credentials == mock.sentinel.credentials + assert isinstance(authed_http.http, urllib3.PoolManager) + + def test_urlopen_no_refresh(self): + credentials = mock.Mock(wraps=CredentialsStub()) + response = ResponseStub() + http = HttpStub([response]) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + + result = authed_http.urlopen("GET", self.TEST_URL) + + assert result == response + assert credentials.before_request.called + assert not credentials.refresh.called + assert http.requests == [ + ("GET", self.TEST_URL, None, {"authorization": "token"}, {}) + ] + + def test_urlopen_refresh(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.OK) + # First request will 401, second request will succeed. + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED), final_response]) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + + authed_http = authed_http.urlopen("GET", "http://example.com") + + assert authed_http == final_response + assert credentials.before_request.call_count == 2 + assert credentials.refresh.called + assert http.requests == [ + ("GET", self.TEST_URL, None, {"authorization": "token"}, {}), + ("GET", self.TEST_URL, None, {"authorization": "token1"}, {}), + ] + + def test_urlopen_no_default_host(self): + credentials = mock.create_autospec(service_account.Credentials) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials) + + authed_http.credentials._create_self_signed_jwt.assert_called_once_with(None) + + def test_urlopen_with_default_host(self): + default_host = "pubsub.googleapis.com" + credentials = mock.create_autospec(service_account.Credentials) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, default_host=default_host + ) + + authed_http.credentials._create_self_signed_jwt.assert_called_once_with( + "https://{}/".format(default_host) + ) + + def test_proxies(self): + http = mock.create_autospec(urllib3.PoolManager) + authed_http = google.auth.transport.urllib3.AuthorizedHttp(None, http=http) + + with authed_http: + pass + + assert http.__enter__.called + assert http.__exit__.called + + authed_http.headers = mock.sentinel.headers + assert authed_http.headers == http.headers + + @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) + def test_configure_mtls_channel_with_callback(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(), http=mock.Mock() + ) + + with pytest.warns(UserWarning): + 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 + 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_metadata( + self, mock_get_client_cert_and_key, mock_make_mutual_tls_http + ): + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials=mock.Mock() + ) + + mock_get_client_cert_and_key.return_value = ( + True, + pytest.public_cert_bytes, + pytest.private_key_bytes, + ) + 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_get_client_cert_and_key.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_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 + ) + def test_configure_mtls_channel_non_mtls( + self, mock_get_client_cert_and_key, mock_make_mutual_tls_http + ): + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials=mock.Mock() + ) + + mock_get_client_cert_and_key.return_value = (False, None, None) + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + is_mtls = authed_http.configure_mtls_channel() + + assert not is_mtls + # If client certificate and key are not found, the transport falls back to + # a standard connection. _is_mtls must be False to reflect this fallback state. + assert authed_http._is_mtls is False + mock_get_client_cert_and_key.assert_called_once() + mock_make_mutual_tls_http.assert_not_called() + + @mock.patch( + "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True + ) + def test_configure_mtls_channel_exceptions(self, mock_get_client_cert_and_key): + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials=mock.Mock() + ) + + mock_get_client_cert_and_key.side_effect = exceptions.ClientCertError() + with pytest.raises(exceptions.MutualTLSChannelError): + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + authed_http.configure_mtls_channel() + assert authed_http._is_mtls is False + + mock_get_client_cert_and_key.side_effect = OSError("Mock file read error") + with pytest.raises(exceptions.MutualTLSChannelError): + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + authed_http.configure_mtls_channel() + assert authed_http._is_mtls is False + + @mock.patch( + "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True + ) + @mock.patch( + "google.auth.transport.urllib3.urllib3.util.ssl_.create_urllib3_context", + autospec=True, + ) + def test_configure_mtls_channel_cert_loading_exceptions( + self, mock_create_urllib3_context, mock_get_client_cert_and_key + ): + import ssl + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials=mock.Mock() + ) + + mock_get_client_cert_and_key.return_value = ( + True, + pytest.public_cert_bytes, + pytest.private_key_bytes, + ) + + for exception_type in [ValueError("error"), ssl.SSLError("error")]: + mock_ctx = mock.Mock() + mock_ctx.load_cert_chain.side_effect = exception_type + mock_create_urllib3_context.return_value = mock_ctx + + with pytest.raises(exceptions.MutualTLSChannelError): + with mock.patch.dict( + os.environ, + {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}, + ): + authed_http.configure_mtls_channel() + assert authed_http._is_mtls is False + + assert not authed_http._is_mtls + + @mock.patch( + "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True + ) + @mock.patch.dict( + os.environ, + { + "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false", + "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false", + "GOOGLE_API_CERTIFICATE_CONFIG": "", + "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", + }, + ) + def test_configure_mtls_channel_without_client_cert_env( + self, get_client_cert_and_key + ): + callback = mock.Mock() + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials=mock.Mock(), http=mock.Mock() + ) + + env_to_patch = { + environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "", + environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE: "", + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG: "", + environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH: "", + } + with mock.patch.dict(os.environ, env_to_patch): + # Test the callback is not called if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. + is_mtls = authed_http.configure_mtls_channel(callback) + assert not is_mtls + callback.assert_not_called() + + # Test ADC client cert is not used if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set. + is_mtls = authed_http.configure_mtls_channel(callback) + assert not is_mtls + get_client_cert_and_key.assert_not_called() + + def test_clear_pool_on_del(self): + http = mock.create_autospec(urllib3.PoolManager) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + mock.sentinel.credentials, http=http + ) + authed_http.__del__() + http.clear.assert_called_with() + + authed_http.http = None + authed_http.__del__() + # Expect it to not crash + + @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) + @mock.patch("google.auth.transport.urllib3._make_default_http", autospec=True) + def test_cert_rotation_when_cert_mismatch_and_mtls_endpoint_used( + self, mock_make_default_http, mock_make_mutual_tls_http + ): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.OK) + + # We simulate the HTTP stub rotation. When mTLS http is created, we return rotated_http. + rotated_http = HttpStub([final_response]) + mock_make_mutual_tls_http.return_value = rotated_http + + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + + old_cert = b"-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n" + + # New certificate and key to simulate rotation. + new_cert = pytest.public_cert_bytes + new_key = pytest.private_key_bytes + # Set _cached_cert to a callable that returns the old certificate. + authed_http._cached_cert = old_cert + authed_http._is_mtls = True + # Mock call_client_cert_callback to return the new certificate. + with mock.patch.object( + google.auth.transport._mtls_helper, + "call_client_cert_callback", + return_value=(new_cert, new_key), + ) as mock_callback: + # mTLS endpoint is used, and client cert env var is true + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + result = authed_http.urlopen( + "GET", "http://example.mtls.googleapis.com" + ) + + # Asserts to verify the behavior. + assert result == final_response + assert credentials.refresh.called + assert credentials.refresh.call_count == 1 + assert mock_callback.called + mock_make_mutual_tls_http.assert_called_once_with(cert=new_cert, key=new_key) + + def test_no_cert_rotation_when_cert_match_and_mtls_endpoint_used(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.UNAUTHORIZED) + http = HttpStub( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.UNAUTHORIZED), + ] + ) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + old_cert = CERT_MOCK_VAL + + new_cert = old_cert + new_key = KEY_MOCK_VAL + # Set _cached_cert to a callable that returns the same certificate. + authed_http._cached_cert = old_cert + authed_http._is_mtls = True + # Mock call_client_cert_callback to return the certificate. + with mock.patch.object( + google.auth.transport._mtls_helper, + "call_client_cert_callback", + return_value=(new_cert, new_key), + ): + # mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + + # Asserts to verify the behavior. + assert credentials.refresh.call_count == 2 + assert result.status == final_response.status + + def test_no_cert_match_check_when_mtls_endpoint_not_used(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.UNAUTHORIZED) + http = HttpStub( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.UNAUTHORIZED), + ] + ) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + authed_http._is_mtls = False + new_cert = CERT_MOCK_VAL + new_key = KEY_MOCK_VAL + + # Mock call_client_cert_callback to return the certificate. + with mock.patch.object( + google.auth.transport._mtls_helper, + "call_client_cert_callback", + return_value=(new_cert, new_key), + ) as mock_callback: + # non-mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.googleapis.com") + + # Asserts to verify the behavior. + assert not mock_callback.called + assert result.status == final_response.status + + def test_no_cert_rotation_when_no_unauthorized_response(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.UPGRADE_REQUIRED) + + # Response is set to code other than 401(Unauthorized). + http = HttpStub([ResponseStub(status=http_client.UPGRADE_REQUIRED)]) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + authed_http._is_mtls = True + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + # mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + assert result.status == final_response.status + assert not credentials.refresh.called + assert credentials.refresh.call_count == 0 + + def test_cert_rotation_failure_raises_error(self): + credentials = mock.Mock(wraps=CredentialsStub()) + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + + old_cert = b"-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n" + + # New certificate and key to simulate rotation. + new_cert = CERT_MOCK_VAL + new_key = KEY_MOCK_VAL + authed_http._cached_cert = old_cert + authed_http._is_mtls = True + + # Mock call_client_cert_callback to return the new certificate. + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=(new_cert, new_key, "old_fingerprint", "new_fingerprint"), + ) as mock_check_params: + with mock.patch.object( + authed_http, + "configure_mtls_channel", + side_effect=Exception("Failed to reconfigure"), + ) as mock_reconfigure: + with pytest.raises(exceptions.MutualTLSChannelError): + authed_http.urlopen("GET", "https://example.mtls.googleapis.com") + + mock_check_params.assert_called_once() + mock_reconfigure.assert_called_once() + credentials.refresh.assert_not_called() + + def test_cert_rotation_check_params_fails(self): + credentials = mock.Mock(wraps=CredentialsStub()) + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + authed_http._is_mtls = True + authed_http._cached_cert = b"cached_cert" + + with mock.patch( + "google.auth.transport.urllib3._mtls_helper.check_parameters_for_unauthorized_response", + side_effect=Exception("check_params failed"), + ) as mock_check_params: + with pytest.raises(Exception, match="check_params failed"): + authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + + mock_check_params.assert_called_once() + credentials.refresh.assert_not_called() + + def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): + """ + Tests that the code can handle a refresh triggered by a status code + other than 401 (UNAUTHORIZED). This covers the 'else' branch of the + 'if response.status_code == http_client.UNAUTHORIZED' check + """ + credentials = mock.Mock(wraps=CredentialsStub()) + # Configure the session to treat 503 (Service Unavailable) as a refreshable error + custom_codes = [http_client.SERVICE_UNAVAILABLE] + + # Return 503 first, then 200 + http = HttpStub( + [ + ResponseStub(status=http_client.SERVICE_UNAVAILABLE), + ResponseStub(status=http_client.OK), + ] + ) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http, refresh_status_codes=custom_codes + ) + + # Enable mTLS to prove it is skipped despite being enabled + authed_http._is_mtls = True + mtls_url = "https://mtls.googleapis.com/test" + + with mock.patch( + "google.auth.transport.urllib3._mtls_helper", autospec=True + ) as mock_helper: + authed_http.urlopen("GET", mtls_url) + + # Assert refresh happened (Outer Check was True) + assert credentials.refresh.called + + # Assert mTLS check logic was SKIPPED (Inner Check was False) + assert not mock_helper.check_parameters_for_unauthorized_response.called + + @mock.patch("google.auth.transport.urllib3._make_mutual_tls_http", autospec=True) + def test_configure_mtls_channel_subsequent_failure(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 fails + 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.side_effect = exceptions.ClientCertError() + + 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 it retains its previous mTLS state and connection pool + 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_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 returns no client certificate (disabled) + 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 mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + is_mtls = 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) + + +class TestAuthorizedHttpMTLSReauth: + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params): + credentials = mock.Mock() + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) + http_obj._is_mtls = True + http_obj._cached_cert = b"cert" + mock_response = mock.Mock() + mock_response.status = http_client.UNAUTHORIZED + http_obj.http.urlopen = mock.Mock(return_value=mock_response) + real_lock = threading.Lock() + http_obj._reauth_lock = real_lock + mock_check_params.return_value = ( + b"new_cert_bytes", + b"new_key_bytes", + "old_fingerprint", + "new_fingerprint", + ) + lock_held_during_call = {"held": False} + + def verify_lock_held(*args, **kwargs): + lock_held_during_call["held"] = real_lock.locked() + + http_obj.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held) + http_obj.request("GET", "https://example.mtls.googleapis.com/") + http_obj.configure_mtls_channel.assert_called() + assert lock_held_during_call["held"] is True + + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + def test_reauth_skipped_when_cert_fingerprint_matches(self, mock_check_params): + credentials = mock.Mock() + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) + http_obj._is_mtls = True + http_obj._cached_cert = b"cert" + mock_response_unauth = mock.Mock() + mock_response_unauth.status = http_client.UNAUTHORIZED + mock_response_ok = mock.Mock() + mock_response_ok.status = http_client.OK + http_obj.http.urlopen = mock.Mock( + side_effect=[mock_response_unauth, mock_response_ok] + ) + mock_check_params.return_value = ( + b"same_cert_bytes", + b"same_key_bytes", + "same_fingerprint", + "same_fingerprint", + ) + http_obj.configure_mtls_channel = mock.Mock() + http_obj.request("GET", "https://example.mtls.googleapis.com/") + http_obj.configure_mtls_channel.assert_not_called()