Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
macastelaz merged 1 commit into from
Aug 21, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 0 additions & 48 deletions packages/google-auth/google/auth/transport/_mtls_helper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
8 changes: 7 additions & 1 deletion packages/google-auth/google/auth/transport/requests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Comment on lines +650 to +656

@gemini-code-assist gemini-code-assist Bot Aug 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Using a simple substring check (prefix in url) on the raw url parameter can raise a TypeError if url is a bytes object or a urllib3.util.Url object. To ensure robust type handling, check for a .url attribute, safely decode bytes to UTF-8, and fall back to string conversion.

                MTLS_URL_PREFIXES = [
                    "mtls.googleapis.com",
                    "mtls.sandbox.googleapis.com",
                ]
                url_str = url.url if hasattr(url, "url") else (url.decode("utf-8") if isinstance(url, bytes) else str(url))
                use_mtls = self.is_mtls and any(
                    prefix in url_str for prefix in MTLS_URL_PREFIXES
                )
References
  1. When parsing or validating URLs that may be passed as urllib3.util.Url objects, bytes, or strings, ensure robust type handling by checking for a .url attribute, safely decoding bytes to UTF-8, and falling back to string conversion to prevent TypeErrors during parsing.

if use_mtls:
(
call_cert_bytes,
Expand Down
6 changes: 5 additions & 1 deletion packages/google-auth/google/auth/transport/urllib3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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])
Comment on lines +412 to +415

@gemini-code-assist gemini-code-assist Bot Aug 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Using a list comprehension inside any() defeats short-circuiting. Additionally, url can be a urllib3.util.Url object, bytes, or str. To prevent TypeError, check for a .url attribute, safely decode bytes to UTF-8, and fall back to string conversion.

Suggested change
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])
use_mtls = False
if self._is_mtls:
MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"]
url_str = url.url if hasattr(url, "url") else (url.decode("utf-8") if isinstance(url, bytes) else str(url))
use_mtls = any(prefix in url_str for prefix in MTLS_URL_PREFIXES)
References
  1. When parsing or validating URLs that may be passed as urllib3.util.Url objects, bytes, or strings, ensure robust type handling by checking for a .url attribute, safely decoding bytes to UTF-8, and falling back to string conversion to prevent TypeErrors during parsing.


# 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()
Expand All @@ -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,
Expand Down
78 changes: 0 additions & 78 deletions packages/google-auth/tests/transport/test__mtls_helper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
53 changes: 8 additions & 45 deletions packages/google-auth/tests/transport/test_requests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Expand All @@ -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: "",
Expand Down Expand Up @@ -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())
Expand All @@ -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
Expand Down
66 changes: 1 addition & 65 deletions packages/google-auth/tests/transport/test_urllib3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Expand All @@ -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()

Expand Down Expand Up @@ -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()
Expand Down
Loading