Skip to content

fix(auth): only trigger mTLS certificate rotation on mTLS endpoints #17928

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
attharva-24 merged 3 commits into from
Jul 29, 2026
+51 −13
Merged

fix(auth): only trigger mTLS certificate rotation on mTLS endpoints #17928

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c725c3a
fix(auth): only trigger mTLS certificate rotation on mTLS endpoints
attharva-24 Jul 28, 2026
d44dfc9
style: format with black
attharva-24 Jul 28, 2026
164b18e
Merge branch 'origin/main' into fix/requests-mtls-url-check
attharva-24 Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion packages/google-auth/google/auth/transport/requests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -647,7 +647,14 @@ 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,
Expand Down
55 changes: 43 additions & 12 deletions packages/google-auth/tests/transport/test_requests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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"

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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"

Expand All @@ -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()
Expand All @@ -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"

Expand All @@ -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()
Expand All @@ -920,22 +921,52 @@ 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

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

# 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()
Expand Down
Loading