Revert "fix(auth): parse hostname for mTLS and PSC endpoint certifica… (#18189)

(#18153)

This reverts commit b642373.

Author: attharva-24

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

@@ -24,7 +24,6 @@
 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
@@ -841,50 +840,3 @@ 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",
-    "p.googleapis.com",
-)
-
-
-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, object]]): 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.rstrip(".").lower()
-    if not hostname:
-        return False
-
-    return hostname in _MTLS_EXACT_HOSTS or hostname.endswith(_MTLS_HOST_SUFFIXES)

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

@@ -647,7 +647,13 @@ def request(
         ):
             # Handle unauthorized permission error(401 status code)
             if response.status_code == http_client.UNAUTHORIZED:
-                use_mtls = self.is_mtls and _mtls_helper.is_mtls_endpoint(url)
+                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,

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

@@ -409,6 +409,11 @@ 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()
@@ -431,7 +436,6 @@ 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

@@ -21,7 +21,6 @@
 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
@@ -1889,80 +1888,3 @@ 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"),
-            "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/",
-            "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):
-        assert _mtls_helper.is_mtls_endpoint(url) is True
-
-    @pytest.mark.parametrize(
-        "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://[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",
-            "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"
-            ),
-            "https://.",
-            "",
-            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,9 +664,6 @@ 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
     )
@@ -680,7 +677,7 @@ def test_configure_mtls_channel_cert_loading_exceptions(
         },
     )
     def test_configure_mtls_channel_without_client_cert_env(
-        self, get_client_cert_and_key, mock_get_cert_config_path
+        self, get_client_cert_and_key
     ):
         env_to_patch = {
             environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "",
@@ -942,7 +939,7 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self):
 
     def test_cert_rotation_skipped_on_non_mtls_url(self):
         """
-        Tests that mTLS cert rotation is skipped on non-mTLS URLs even if
+        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())
@@ -953,56 +950,22 @@ def test_cert_rotation_skipped_on_non_mtls_url(self):
                 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(self.TEST_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",
-        ) as mock_check_params:
-            authed_session.request("GET", non_mtls_url)
+        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_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
+            assert not mock_helper.check_parameters_for_unauthorized_response.called
 
     def test_configure_mtls_channel_subsequent_failure(self):
         # 1. Setup successful mTLS configuration

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

@@ -385,9 +385,6 @@ 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
     )
@@ -401,7 +398,7 @@ def test_configure_mtls_channel_cert_loading_exceptions(
         },
     )
     def test_configure_mtls_channel_without_client_cert_env(
-        self, get_client_cert_and_key, mock_get_cert_config_path
+        self, get_client_cert_and_key
     ):
         callback = mock.Mock()
 
@@ -658,67 +655,6 @@ 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 non-mTLS URLs 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),
-            ]
-        )
-        non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com"
-        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()