From c725c3a6d7ac50386c5481007d0acb7e4faa6d92 Mon Sep 17 00:00:00 2001 From: Atharva Date: Tue, 28 Jul 2026 20:13:18 +0000 Subject: [PATCH 1/2] fix(auth): only trigger mTLS certificate rotation on mTLS endpoints --- .../google/auth/transport/requests.py | 4 +- .../tests/transport/test_requests.py | 55 +++++++++++++++---- 2 files changed, 46 insertions(+), 13 deletions(-) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 12f4dda8e084..7c581754fc81 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -647,7 +647,9 @@ def request( ): # Handle unauthorized permission error(401 status code) if response.status_code == http_client.UNAUTHORIZED: - if self.is_mtls: + MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"] + use_mtls = self.is_mtls and any([prefix in url for prefix in MTLS_URL_PREFIXES]) + if use_mtls: ( call_cert_bytes, call_key_bytes, diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 5aba3772132e..2ca1922494ef 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -213,6 +213,7 @@ def make_response(status=http_client.OK, data=None): class TestAuthorizedSession(object): TEST_URL = "http://example.com/" + MTLS_TEST_URL = "https://example.mtls.googleapis.com/" def test_constructor(self): authed_session = google.auth.transport.requests.AuthorizedSession( @@ -727,7 +728,7 @@ def test_cert_rotation_when_cert_mismatch_and_mtls_enabled(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount(self.MTLS_TEST_URL, adapter) old_cert = b"-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n" @@ -745,7 +746,7 @@ def test_cert_rotation_when_cert_mismatch_and_mtls_enabled(self): "call_client_cert_callback", return_value=(new_cert, new_key), ) as mock_callback: - result = authed_session.request("GET", self.TEST_URL) + result = authed_session.request("GET", self.MTLS_TEST_URL) # Asserts to verify the behavior. assert mock_callback.called @@ -766,7 +767,7 @@ def test_no_cert_rotation_when_cert_match_and_mTLS_enabled(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount(self.MTLS_TEST_URL, adapter) authed_session._is_mtls = True old_cert = CERT_MOCK_VAL @@ -784,7 +785,7 @@ def test_no_cert_rotation_when_cert_match_and_mTLS_enabled(self): "call_client_cert_callback", return_value=(new_cert, new_key), ): - result = authed_session.request("GET", self.TEST_URL) + result = authed_session.request("GET", self.MTLS_TEST_URL) # Asserts to verify the behavior. assert credentials.refresh.call_count == 2 @@ -832,11 +833,11 @@ def test_no_cert_rotation_when_no_unauthorized_response(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount(self.MTLS_TEST_URL, adapter) authed_session._is_mtls = True - result = authed_session.request("GET", self.TEST_URL) + result = authed_session.request("GET", self.MTLS_TEST_URL) assert result.status_code == final_response.status_code # Asserts to verify the behavior. @@ -851,7 +852,7 @@ def test_cert_rotation_failure_raises_error(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount(self.MTLS_TEST_URL, adapter) old_cert = b"-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n" @@ -873,7 +874,7 @@ def test_cert_rotation_failure_raises_error(self): side_effect=Exception("Failed to reconfigure"), ): with pytest.raises(exceptions.MutualTLSChannelError): - authed_session.request("GET", self.TEST_URL) + authed_session.request("GET", self.MTLS_TEST_URL) # Assert to verify behavior credentials.refresh.assert_not_called() @@ -885,7 +886,7 @@ def test_cert_rotation_check_params_fails(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount(self.MTLS_TEST_URL, adapter) authed_session._is_mtls = True authed_session._cached_cert = b"cached_cert" @@ -894,7 +895,7 @@ def test_cert_rotation_check_params_fails(self): side_effect=Exception("check_params failed"), ) as mock_check_params: with pytest.raises(Exception, match="check_params failed"): - authed_session.request("GET", self.TEST_URL) + authed_session.request("GET", self.MTLS_TEST_URL) mock_check_params.assert_called_once() credentials.refresh.assert_not_called() @@ -920,7 +921,7 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_status_codes=custom_refresh_codes ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount(self.MTLS_TEST_URL, adapter) # Enable mTLS to prove it is skipped despite being enabled authed_session._is_mtls = True @@ -928,7 +929,7 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): with mock.patch( "google.auth.transport.requests._mtls_helper", autospec=True ) as mock_helper: - authed_session.request("GET", self.TEST_URL) + authed_session.request("GET", self.MTLS_TEST_URL) # Assert refresh happened (Outer Check was True) assert credentials.refresh.called @@ -936,6 +937,36 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): # Assert mTLS check logic was SKIPPED (Inner Check was False) assert not mock_helper.check_parameters_for_unauthorized_response.called + def test_cert_rotation_skipped_on_non_mtls_url(self): + """ + Tests that mTLS cert rotation is skipped on a non-mTLS URL even if + mTLS is enabled and an UNAUTHORIZED (401) response is received. + """ + credentials = mock.Mock(wraps=CredentialsStub()) + # First request will 401, second request will succeed. + adapter = AdapterStub( + [ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ] + ) + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(self.TEST_URL, adapter) + authed_session._is_mtls = True + + with mock.patch( + "google.auth.transport.requests._mtls_helper", autospec=True + ) as mock_helper: + authed_session.request("GET", self.TEST_URL) + + # Assert refresh happened + assert credentials.refresh.called + + # Assert mTLS check logic was SKIPPED + assert not mock_helper.check_parameters_for_unauthorized_response.called + def test_configure_mtls_channel_subsequent_failure(self): # 1. Setup successful mTLS configuration mock_callback = mock.Mock() From d44dfc9f8e97d649e53b82c8ea4219754b11cb1f Mon Sep 17 00:00:00 2001 From: Atharva Date: Tue, 28 Jul 2026 20:57:52 +0000 Subject: [PATCH 2/2] style: format with black --- packages/google-auth/google/auth/transport/requests.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 7c581754fc81..cb057eb21ac9 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -647,8 +647,13 @@ def request( ): # Handle unauthorized permission error(401 status code) if response.status_code == http_client.UNAUTHORIZED: - MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"] - use_mtls = self.is_mtls and any([prefix in url for prefix in MTLS_URL_PREFIXES]) + MTLS_URL_PREFIXES = [ + "mtls.googleapis.com", + "mtls.sandbox.googleapis.com", + ] + use_mtls = self.is_mtls and any( + [prefix in url for prefix in MTLS_URL_PREFIXES] + ) if use_mtls: ( call_cert_bytes,