diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 7779c484c713..92272c243c3e 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/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,50 @@ 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 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..f1d096ff64cd 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,80 @@ 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 2ca1922494ef..c106a87f08fb 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: "", @@ -939,7 +942,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 a non-mTLS URL 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()) @@ -950,22 +953,56 @@ 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(self.TEST_URL, adapter) + authed_session.mount("https://", 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..0fbee087e11f 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,67 @@ 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()