Skip to content

fix(auth): parse hostname for mTLS and PSC endpoint certificate rotat… #18153

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

Open
attharva-24 wants to merge 5 commits into
base: main
Choose a base branch
from
Open

fix(auth): parse hostname for mTLS and PSC endpoint certificate rotat… #18153

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b629135
fix(auth): parse hostname for mTLS and PSC endpoint certificate rotat…
attharva-24 Aug 19, 2026
6f85765
fix(auth): strip trailing root dot in hostname for FQDNs
attharva-24 Aug 20, 2026
a9eb0be
test(auth): add test cases for explicit port numbers, queries, and fr…
attharva-24 Aug 21, 2026
9dbd745
fix(auth): support bare PSC domain and add IPv6 unit tests
attharva-24 Aug 21, 2026
f20b946
style(auth): replace Any with object in is_mtls_endpoint type annotat…
attharva-24 Aug 21, 2026
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: 48 additions & 0 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,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
Expand Down Expand Up @@ -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
Comment on lines +867 to +881

@gemini-code-assist gemini-code-assist Bot Aug 19, 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.

medium

In urllib3, the url parameter passed to urlopen can be a urllib3.util.Url object (or other string-like/URL objects) rather than a plain str or bytes. Currently, passing a urllib3.util.Url object to is_mtls_endpoint will cause urlsplit(url) to raise a TypeError, which is caught and results in returning False—even if the object represents a valid mTLS endpoint.

To prevent this and ensure robust compatibility with urllib3's native URL objects, we should check if the input has a .url attribute (which urllib3.util.Url exposes as a property returning the string representation) or fall back to converting it to a string.

Suggested change
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 url:
return False
if isinstance(url, bytes):
try:
url = url.decode("utf-8")
except (UnicodeDecodeError, AttributeError):
return False
elif not isinstance(url, str):
if hasattr(url, "url"):
url = url.url
else:
try:
url = str(url)
except Exception:
return False
try:
hostname = urlsplit(url).hostname
except (ValueError, TypeError, AttributeError):
return False
References
  1. Adheres to the defensive programming guidelines in Section 2 of the Repository Style Guide by validating and normalizing the input type before performing operations that could raise TypeErrors. (link)
  2. Do not replace historical graceful fallback behaviors (such as returning False/falling back to standard TLS) with exceptions if doing so would introduce breaking changes for downstream users and violate backwards compatibility.

@parthea parthea Aug 19, 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.

@attharva-24 PTAL

@attharva-24 attharva-24 Aug 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Addressed in the latest commit!

is_mtls_endpoint now checks for .url on urllib3.util.Url (and other URL objects), supports bytes inputs with safe UTF-8 decoding, and falls back to string conversion. Added unit tests covering urllib3.util.Url, bytes, and str inputs across standard, PSC, and regional mTLS endpoints.


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: 1 addition & 7 deletions 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,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,
Expand Down
6 changes: 1 addition & 5 deletions 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,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()
Expand All @@ -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,
Expand Down
78 changes: 78 additions & 0 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,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
Expand Down Expand Up @@ -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",
[

@macastelaz macastelaz Aug 20, 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.

This lacks examples with explicit port numbers (example: https://pubsub.mtls.googleapis.com:443/v1) and queries and fragments (e.g. "https://pubsub.mtls.googleapis.com/v1/projects?pageSize=10#frag")

@attharva-24 attharva-24 Aug 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Added unit test cases in TestIsMtlsEndpoint covering explicit port numbers (:443, :8443), query parameters, URL fragments, and combined port+query+fragment permutations for standard mTLS, PSC, and non-mTLS endpoints.

a9eb0be

"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",
[

@macastelaz macastelaz Aug 20, 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.

Consider adding a bare PSC case (example "https://p.googleapis.com").

@macastelaz macastelaz Aug 20, 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.

Additionally a case like https://[2001:db8::1]:443/mtls.googleapis.com would be good to demonstrated handling of IPv6 syntax handling

@attharva-24 attharva-24 Aug 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Added "p.googleapis.com" to _MTLS_EXACT_HOSTS so bare apex PSC domains are recognized as mTLS endpoints, along with unit test cases for "https://p.googleapis.com", ports, and trailing root dots.

Added IPv6 test cases (https://[2001:db8::1]:443/mtls.googleapis.com and https://[::1]:8443/mtls.googleapis.com) to confirm that bracketed IPv6 host syntax is handled properly.

9dbd745

"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
65 changes: 56 additions & 9 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,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
)
Expand All @@ -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: "",
Expand Down Expand Up @@ -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(

@macastelaz macastelaz Aug 20, 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.

I think this is redundant - these transports are no longer responsible for mtls checks themselves and the thing under test here shouldn't be if various forms on non-mtls endpoints are detected correctly (that is already covered in the new mtls_helper tests). Instead, I'd suggest just covering one example of mtls and one example of non mtls here to cover the requests logic specifically. Same for urllib3

@macastelaz macastelaz 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.

Looks like just this comment is pending and then I'll take one more look.

"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())
Expand All @@ -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
Expand Down
75 changes: 74 additions & 1 deletion 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,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
)
Expand All @@ -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()

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