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

Author: attharva-24

packages/google-auth/google/auth/transport/_mtls_helper.py

@@ -24,6 +24,7 @@
 import sys
 import tempfile
 from typing import 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,42 @@ 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]]) -> bool:
+    """Checks if the given URL corresponds to an mTLS or Private Service Connect (PSC) endpoint.
+
+    Args:
+        url (Optional[Union[str, bytes]]): The request URL.
+
+    Returns:
+        bool: True if the URL targets an mTLS or PSC endpoint, False otherwise.
+    """
+    if not url:
+        return False
+    if isinstance(url, bytes):
+        try:
+            url = url.decode("utf-8")
+        except (UnicodeDecodeError, AttributeError):
+            return False
+    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)

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,

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,

packages/google-auth/tests/transport/test__mtls_helper.py

@@ -1888,3 +1888,54 @@ 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",
+        ],
+    )
+    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",
+            "",
+            None,
+            123,
+            "not a url",
+        ],
+    )
+    def test_is_mtls_endpoint_false(self, url):
+        assert _mtls_helper.is_mtls_endpoint(url) is False

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

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()