diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 92272c243c3e..7779c484c713 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/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) diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 73bb7e719f98..822cf687f5d0 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/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, diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index 1a529d3b766e..18e6128e03bd 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/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, diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index f1d096ff64cd..e9bb62db2133 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/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 diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index c106a87f08fb..2ca1922494ef 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/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 diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index 0fbee087e11f..e1c92dbebc2c 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/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()