From f11e33d97563d5996910174fdc76d11f7fd17df7 Mon Sep 17 00:00:00 2001 From: Atharva Date: Wed, 19 Aug 2026 06:44:57 +0000 Subject: [PATCH 1/6] fix(auth): parse hostname for mTLS and PSC endpoint certificate rotation (#18147) * Isolate hostname using urllib.parse.urlsplit in _mtls_helper.is_mtls_endpoint * Eliminate false positives on non-mTLS URLs containing mtls substrings in paths/queries * Add support for Private Service Connect (*.p.googleapis.com) custom mTLS endpoints * Update AuthorizedSession and AuthorizedHttp to use shared is_mtls_endpoint helper * Add comprehensive unit tests in test__mtls_helper, test_requests, and test_urllib3 Fixes #18147 Follow-up to #17928 --- .../google/auth/transport/_mtls_helper.py | 46 +++++++++++- .../google/auth/transport/requests.py | 8 +- .../google/auth/transport/urllib3.py | 6 +- .../tests/transport/test__mtls_helper.py | 58 ++++++++++++++ .../tests/transport/test_requests.py | 65 +++++++++++++--- .../tests/transport/test_urllib3.py | 75 ++++++++++++++++++- 6 files changed, 235 insertions(+), 23 deletions(-) diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 7779c484c713..8a1995fe6526 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -23,7 +23,8 @@ import subprocess import sys import tempfile -from typing import cast, Generator, List, Optional, Tuple, Union +from typing import Any, cast, Generator, List, Optional, Tuple, Union +from urllib.parse import urlsplit from google.auth import _agent_identity_utils from google.auth import _cloud_sdk @@ -840,3 +841,46 @@ def call_client_cert_callback(): generate_encrypted_key=True ) return cert_bytes, key_bytes + + +_MTLS_HOST_SUFFIXES = ( + ".mtls.googleapis.com", + ".mtls.sandbox.googleapis.com", + ".p.googleapis.com", +) +_MTLS_EXACT_HOSTS = ( + "mtls.googleapis.com", + "mtls.sandbox.googleapis.com", +) + + +def is_mtls_endpoint(url: Optional[Union[str, bytes, Any]]) -> bool: + """Checks if the given URL corresponds to an mTLS or Private Service Connect (PSC) endpoint. + + Args: + url (Optional[Union[str, bytes, Any]]): The request URL. + + Returns: + bool: True if the URL targets an mTLS or PSC endpoint, False otherwise. + """ + if not url: + return False + if hasattr(url, "url") and isinstance(url.url, (str, bytes)): + url = url.url + if isinstance(url, bytes): + try: + url = url.decode("utf-8") + except (UnicodeDecodeError, AttributeError): + return False + elif not isinstance(url, str): + url = str(url) + try: + hostname = urlsplit(url).hostname + except (ValueError, TypeError, AttributeError): + return False + + if not hostname: + return False + + hostname = hostname.lower() + return hostname in _MTLS_EXACT_HOSTS or hostname.endswith(_MTLS_HOST_SUFFIXES) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 822cf687f5d0..73bb7e719f98 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -647,13 +647,7 @@ 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 - ) + use_mtls = self.is_mtls and _mtls_helper.is_mtls_endpoint(url) if use_mtls: ( call_cert_bytes, diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index 18e6128e03bd..1a529d3b766e 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -409,11 +409,6 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): if headers is None: headers = self.headers - use_mtls = False - if self._is_mtls: - MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"] - use_mtls = any([prefix in url for prefix in MTLS_URL_PREFIXES]) - # Make a copy of the headers. They will be modified by the credentials # and we want to pass the original headers if we recurse. request_headers = headers.copy() @@ -436,6 +431,7 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): and _credential_refresh_attempt < self._max_refresh_attempts ): if response.status == http_client.UNAUTHORIZED: + use_mtls = self._is_mtls and _mtls_helper.is_mtls_endpoint(url) if use_mtls: ( call_cert_bytes, diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index e9bb62db2133..d3b40c43ea4d 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -21,6 +21,7 @@ from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec import pytest # type: ignore +import urllib3.util from google.auth import environment_vars, exceptions from google.auth.transport import _mtls_helper @@ -1888,3 +1889,60 @@ def test_remove_oserror_ignored( mock_fh.flush.assert_called_once() mock_fsync.assert_called_once() mock_remove.assert_called_once_with("/path/to/secret") + + +class TestIsMtlsEndpoint(object): + @pytest.mark.parametrize( + "url", + [ + "https://mtls.googleapis.com", + "https://mtls.googleapis.com/", + "https://mtls.googleapis.com/v1/projects", + "https://mtls.sandbox.googleapis.com", + "https://mtls.sandbox.googleapis.com/v1/projects", + "https://pubsub.mtls.googleapis.com", + "https://pubsub.mtls.googleapis.com/v1/projects/my-project", + "https://storage.mtls.sandbox.googleapis.com/b/my-bucket", + "https://my-service.us-east1.rep.mtls.googleapis.com/v1", + "https://my-service.us-east1.rep.mtls.sandbox.googleapis.com/v1", + "https://storage.p.googleapis.com/b/my-bucket", + "https://my-custom-endpoint.p.googleapis.com/v1", + "https://my-service.us-east1.p.googleapis.com/v1", + "HTTP://PUBSUB.MTLS.GOOGLEAPIS.COM/V1", + b"https://pubsub.mtls.googleapis.com", + b"https://storage.p.googleapis.com/b/my-bucket", + urllib3.util.parse_url("https://pubsub.mtls.googleapis.com/v1"), + urllib3.util.parse_url("https://storage.p.googleapis.com/b/my-bucket"), + ], + ) + def test_is_mtls_endpoint_true(self, url): + assert _mtls_helper.is_mtls_endpoint(url) is True + + @pytest.mark.parametrize( + "url", + [ + "https://storage.googleapis.com", + "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", + "https://logging.googleapis.com/v2/entries?filter=mtls.sandbox.googleapis.com", + "https://logging.googleapis.com/v2/entries?filter=service.p.googleapis.com", + "https://example.com/mtls.googleapis.com", + "https://fake-mtls.googleapis.com.attacker.com/v1", + "https://fake-p.googleapis.com.attacker.com/v1", + "http://localhost:8080/", + "http://localhost:8080/mtls.googleapis.com", + b"https://storage.googleapis.com", + b"https://storage.googleapis.com/bucket/mtls.googleapis.com", + b"\xff\xfeinvalid", + urllib3.util.parse_url("https://storage.googleapis.com/b/my-bucket"), + urllib3.util.parse_url( + "https://storage.googleapis.com/bucket/mtls.googleapis.com" + ), + "", + None, + 123, + "not a url", + ], + ) + def test_is_mtls_endpoint_false(self, url): + assert _mtls_helper.is_mtls_endpoint(url) is False diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 2ca1922494ef..12c107d41e37 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -664,6 +664,9 @@ def test_configure_mtls_channel_cert_loading_exceptions( assert not auth_session.is_mtls + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", return_value=None + ) @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) @@ -677,7 +680,7 @@ def test_configure_mtls_channel_cert_loading_exceptions( }, ) def test_configure_mtls_channel_without_client_cert_env( - self, get_client_cert_and_key + self, get_client_cert_and_key, mock_get_cert_config_path ): env_to_patch = { environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "", @@ -937,9 +940,19 @@ 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): + @pytest.mark.parametrize( + "non_mtls_url", + [ + "http://example.com/", + "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", + "https://example.com/mtls.sandbox.googleapis.com", + ], + ) + def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): """ - Tests that mTLS cert rotation is skipped on a non-mTLS URL even if + Tests that mTLS cert rotation is skipped on non-mTLS URLs (including + those containing mTLS substrings in paths/query parameters) even if mTLS is enabled and an UNAUTHORIZED (401) response is received. """ credentials = mock.Mock(wraps=CredentialsStub()) @@ -953,19 +966,53 @@ def test_cert_rotation_skipped_on_non_mtls_url(self): authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(self.TEST_URL, adapter) + authed_session.mount("https://", adapter) + authed_session.mount("http://", adapter) authed_session._is_mtls = True + authed_session._cached_cert = b"cached_cert" - with mock.patch( - "google.auth.transport.requests._mtls_helper", autospec=True - ) as mock_helper: - authed_session.request("GET", self.TEST_URL) + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + ) as mock_check_params: + authed_session.request("GET", non_mtls_url) # Assert refresh happened assert credentials.refresh.called # Assert mTLS check logic was SKIPPED - assert not mock_helper.check_parameters_for_unauthorized_response.called + assert not mock_check_params.called + + def test_cert_rotation_triggered_on_psc_url(self): + """ + Tests that mTLS cert rotation IS triggered on a Private Service Connect + (PSC) mTLS endpoint when an UNAUTHORIZED (401) response is received. + """ + credentials = mock.Mock(wraps=CredentialsStub()) + adapter = AdapterStub( + [ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ] + ) + psc_url = "https://storage.p.googleapis.com/b/my-bucket" + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(psc_url, adapter) + authed_session._is_mtls = True + authed_session._cached_cert = b"cached_cert" + + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=(b"new_cert", b"new_key", "old_fp", "old_fp"), + ) as mock_check_params: + authed_session.request("GET", psc_url) + + # Assert mTLS check logic was called on PSC endpoint + mock_check_params.assert_called_once() + assert credentials.refresh.called def test_configure_mtls_channel_subsequent_failure(self): # 1. Setup successful mTLS configuration diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index e1c92dbebc2c..f1d28a070a2b 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -385,6 +385,9 @@ def test_configure_mtls_channel_cert_loading_exceptions( assert not authed_http._is_mtls + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", return_value=None + ) @mock.patch( "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True ) @@ -398,7 +401,7 @@ def test_configure_mtls_channel_cert_loading_exceptions( }, ) def test_configure_mtls_channel_without_client_cert_env( - self, get_client_cert_and_key + self, get_client_cert_and_key, mock_get_cert_config_path ): callback = mock.Mock() @@ -655,6 +658,76 @@ 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 + @pytest.mark.parametrize( + "non_mtls_url", + [ + "http://example.com/", + "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", + "https://example.com/mtls.sandbox.googleapis.com", + ], + ) + def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): + """ + Tests that mTLS cert rotation is skipped on non-mTLS URLs (including + those containing mTLS substrings in paths/query parameters) even if + mTLS is enabled and an UNAUTHORIZED (401) response is received. + """ + credentials = mock.Mock(wraps=CredentialsStub()) + http = HttpStub( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.OK), + ] + ) + 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.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + ) as mock_check_params: + authed_http.urlopen("GET", non_mtls_url) + + # Assert refresh happened + assert credentials.refresh.called + + # Assert mTLS check logic was SKIPPED + assert not mock_check_params.called + + def test_cert_rotation_triggered_on_psc_url(self): + """ + Tests that mTLS cert rotation IS triggered on a Private Service Connect + (PSC) mTLS endpoint when an UNAUTHORIZED (401) response is received. + """ + credentials = mock.Mock(wraps=CredentialsStub()) + http = HttpStub( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.OK), + ] + ) + psc_url = "https://storage.p.googleapis.com/b/my-bucket" + 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.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=(b"new_cert", b"new_key", "old_fp", "old_fp"), + ) as mock_check_params: + authed_http.urlopen("GET", psc_url) + + # Assert mTLS check logic was called on PSC endpoint + mock_check_params.assert_called_once() + assert credentials.refresh.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() From e8637105c8b687328479130bb9914033c1dfe338 Mon Sep 17 00:00:00 2001 From: Atharva Date: Thu, 20 Aug 2026 23:51:44 +0000 Subject: [PATCH 2/6] fix(auth): strip trailing root dot in hostname for FQDNs --- packages/google-auth/google/auth/transport/_mtls_helper.py | 5 ++++- packages/google-auth/tests/transport/test__mtls_helper.py | 5 +++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 8a1995fe6526..d1af6a167e48 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -882,5 +882,8 @@ def is_mtls_endpoint(url: Optional[Union[str, bytes, Any]]) -> bool: if not hostname: return False - hostname = hostname.lower() + hostname = hostname.rstrip(".").lower() + if not hostname: + return False + return hostname in _MTLS_EXACT_HOSTS or hostname.endswith(_MTLS_HOST_SUFFIXES) diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index d3b40c43ea4d..62472218f168 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -1913,6 +1913,9 @@ class TestIsMtlsEndpoint(object): b"https://storage.p.googleapis.com/b/my-bucket", urllib3.util.parse_url("https://pubsub.mtls.googleapis.com/v1"), urllib3.util.parse_url("https://storage.p.googleapis.com/b/my-bucket"), + "https://pubsub.mtls.googleapis.com.", + "https://storage.p.googleapis.com./b/my-bucket", + "https://mtls.googleapis.com.", ], ) def test_is_mtls_endpoint_true(self, url): @@ -1922,6 +1925,7 @@ def test_is_mtls_endpoint_true(self, url): "url", [ "https://storage.googleapis.com", + "https://storage.googleapis.com.", "https://storage.googleapis.com/bucket/mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.sandbox.googleapis.com", @@ -1938,6 +1942,7 @@ def test_is_mtls_endpoint_true(self, url): urllib3.util.parse_url( "https://storage.googleapis.com/bucket/mtls.googleapis.com" ), + "https://.", "", None, 123, From c4fc118d56a2e77133952d3004f9416127e7d330 Mon Sep 17 00:00:00 2001 From: Atharva Date: Fri, 21 Aug 2026 00:16:42 +0000 Subject: [PATCH 3/6] test(auth): add test cases for explicit port numbers, queries, and fragments --- .../google-auth/tests/transport/test__mtls_helper.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index 62472218f168..8b92ff0e79a2 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -1916,6 +1916,13 @@ class TestIsMtlsEndpoint(object): "https://pubsub.mtls.googleapis.com.", "https://storage.p.googleapis.com./b/my-bucket", "https://mtls.googleapis.com.", + "https://pubsub.mtls.googleapis.com:443/v1", + "https://pubsub.mtls.googleapis.com:8443/v1", + "https://storage.p.googleapis.com:443/b/my-bucket", + "https://pubsub.mtls.googleapis.com/v1/projects?pageSize=10#frag", + "https://pubsub.mtls.googleapis.com:443/v1/projects?pageSize=10&filter=foo#frag", + "https://storage.p.googleapis.com:443/b/my-bucket?param=1#section", + "https://mtls.googleapis.com:443/", ], ) def test_is_mtls_endpoint_true(self, url): @@ -1926,6 +1933,8 @@ def test_is_mtls_endpoint_true(self, url): [ "https://storage.googleapis.com", "https://storage.googleapis.com.", + "https://storage.googleapis.com:443/b/my-bucket", + "https://storage.googleapis.com:443/bucket/mtls.googleapis.com?pageSize=10#frag", "https://storage.googleapis.com/bucket/mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.sandbox.googleapis.com", From 07f4549944c5988d8dfbf07ef85831fbb9b19c4b Mon Sep 17 00:00:00 2001 From: Atharva Date: Fri, 21 Aug 2026 00:39:14 +0000 Subject: [PATCH 4/6] fix(auth): support bare PSC domain and add IPv6 unit tests --- packages/google-auth/google/auth/transport/_mtls_helper.py | 1 + packages/google-auth/tests/transport/test__mtls_helper.py | 6 ++++++ 2 files changed, 7 insertions(+) diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index d1af6a167e48..5aecc34d74c0 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -851,6 +851,7 @@ def call_client_cert_callback(): _MTLS_EXACT_HOSTS = ( "mtls.googleapis.com", "mtls.sandbox.googleapis.com", + "p.googleapis.com", ) diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index 8b92ff0e79a2..f1d096ff64cd 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -1923,6 +1923,10 @@ class TestIsMtlsEndpoint(object): "https://pubsub.mtls.googleapis.com:443/v1/projects?pageSize=10&filter=foo#frag", "https://storage.p.googleapis.com:443/b/my-bucket?param=1#section", "https://mtls.googleapis.com:443/", + "https://p.googleapis.com", + "https://p.googleapis.com/", + "https://p.googleapis.com:443/v1", + "https://p.googleapis.com.", ], ) def test_is_mtls_endpoint_true(self, url): @@ -1936,6 +1940,8 @@ def test_is_mtls_endpoint_true(self, url): "https://storage.googleapis.com:443/b/my-bucket", "https://storage.googleapis.com:443/bucket/mtls.googleapis.com?pageSize=10#frag", "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://[2001:db8::1]:443/mtls.googleapis.com", + "https://[::1]:8443/mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.sandbox.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=service.p.googleapis.com", From 615ae8be72b399822ac149598a3848a135ffd23b Mon Sep 17 00:00:00 2001 From: Atharva Date: Fri, 21 Aug 2026 01:00:01 +0000 Subject: [PATCH 5/6] style(auth): replace Any with object in is_mtls_endpoint type annotations --- packages/google-auth/google/auth/transport/_mtls_helper.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 5aecc34d74c0..92272c243c3e 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -23,7 +23,7 @@ import subprocess import sys import tempfile -from typing import Any, cast, Generator, List, Optional, Tuple, Union +from typing import cast, Generator, List, Optional, Tuple, Union from urllib.parse import urlsplit from google.auth import _agent_identity_utils @@ -855,11 +855,11 @@ def call_client_cert_callback(): ) -def is_mtls_endpoint(url: Optional[Union[str, bytes, Any]]) -> bool: +def is_mtls_endpoint(url: Optional[Union[str, bytes, object]]) -> bool: """Checks if the given URL corresponds to an mTLS or Private Service Connect (PSC) endpoint. Args: - url (Optional[Union[str, bytes, Any]]): The request URL. + url (Optional[Union[str, bytes, object]]): The request URL. Returns: bool: True if the URL targets an mTLS or PSC endpoint, False otherwise. From 2eb5fe344a986efd10fa793f7a72a3862acab78e Mon Sep 17 00:00:00 2001 From: Atharva Date: Fri, 21 Aug 2026 16:21:03 +0000 Subject: [PATCH 6/6] test(auth): simplify transport rotation tests to single mTLS/non-mTLS examples --- .../google-auth/tests/transport/test_requests.py | 16 +++------------- .../google-auth/tests/transport/test_urllib3.py | 15 +++------------ 2 files changed, 6 insertions(+), 25 deletions(-) diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 12c107d41e37..c106a87f08fb 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -940,19 +940,9 @@ 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 - @pytest.mark.parametrize( - "non_mtls_url", - [ - "http://example.com/", - "https://storage.googleapis.com/bucket/mtls.googleapis.com", - "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", - "https://example.com/mtls.sandbox.googleapis.com", - ], - ) - def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): + def test_cert_rotation_skipped_on_non_mtls_url(self): """ - Tests that mTLS cert rotation is skipped on non-mTLS URLs (including - those containing mTLS substrings in paths/query parameters) even if + Tests that mTLS cert rotation is skipped on non-mTLS URLs even if mTLS is enabled and an UNAUTHORIZED (401) response is received. """ credentials = mock.Mock(wraps=CredentialsStub()) @@ -963,11 +953,11 @@ def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): make_response(status=http_client.OK), ] ) + non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com" authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) authed_session.mount("https://", adapter) - authed_session.mount("http://", adapter) authed_session._is_mtls = True authed_session._cached_cert = b"cached_cert" diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index f1d28a070a2b..0fbee087e11f 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -658,19 +658,9 @@ 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 - @pytest.mark.parametrize( - "non_mtls_url", - [ - "http://example.com/", - "https://storage.googleapis.com/bucket/mtls.googleapis.com", - "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", - "https://example.com/mtls.sandbox.googleapis.com", - ], - ) - def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): + def test_cert_rotation_skipped_on_non_mtls_url(self): """ - Tests that mTLS cert rotation is skipped on non-mTLS URLs (including - those containing mTLS substrings in paths/query parameters) even if + Tests that mTLS cert rotation is skipped on non-mTLS URLs even if mTLS is enabled and an UNAUTHORIZED (401) response is received. """ credentials = mock.Mock(wraps=CredentialsStub()) @@ -680,6 +670,7 @@ def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): ResponseStub(status=http_client.OK), ] ) + non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com" authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, http=http )