fix(auth): parse hostname for mTLS and PSC endpoint certificate rotat… - #18153
fix(auth): parse hostname for mTLS and PSC endpoint certificate rotat…#18153attharva-24 wants to merge 1 commit into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request centralizes and improves mTLS endpoint detection by introducing the is_mtls_endpoint helper function in _mtls_helper.py, replacing previous substring-based checks in both the requests and urllib3 transports. It also adds comprehensive unit tests to verify the new endpoint detection and cert rotation logic. The review feedback highlights a potential TypeError in is_mtls_endpoint when handling bytes URLs, as calling endswith with string suffixes on a bytes hostname outside the try-except block will raise an exception. Decoding bytes inputs to str at the start of the function is recommended to ensure robust error handling.
e33b6c9 to
f217bfc
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request refactors the mTLS endpoint detection logic by introducing a centralized is_mtls_endpoint helper in _mtls_helper.py and updating both requests and urllib3 transports to use it. It also adds comprehensive unit tests to verify the new helper and ensure cert rotation is skipped on non-mTLS URLs. The review feedback suggests improving the robustness of is_mtls_endpoint by handling non-string/non-bytes URL objects (such as urllib3.util.Url) to prevent them from being incorrectly classified as non-mTLS endpoints due to caught TypeErrors.
| 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 |
There was a problem hiding this comment.
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.
| 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
- 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)
- 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.
There was a problem hiding this comment.
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.
…ion (googleapis#18147) * Isolate hostname using urllib.parse.urlsplit in _mtls_helper.is_mtls_endpoint * Eliminate false positives on non-mTLS URLs containing mtls substrings in paths/queries * Add support for Private Service Connect (*.p.googleapis.com) custom mTLS endpoints * Update AuthorizedSession and AuthorizedHttp to use shared is_mtls_endpoint helper * Add comprehensive unit tests in test__mtls_helper, test_requests, and test_urllib3 Fixes googleapis#18147 Follow-up to googleapis#17928
f217bfc to
7fecb67
Compare
| if not hostname: | ||
| return False | ||
|
|
||
| hostname = hostname.lower() |
There was a problem hiding this comment.
nit: I believe technically FQDNs can end with a trailing root dot (e.g. "https://pubsub.mtls.googleapis.com." is technically valid - although not expected in the wild). For complete coverage, perhaps we should also strip "." (e.g. hostname = hostname.rstrip(".").lower();
| class TestIsMtlsEndpoint(object): | ||
| @pytest.mark.parametrize( | ||
| "url", | ||
| [ |
There was a problem hiding this comment.
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")
|
|
||
| @pytest.mark.parametrize( | ||
| "url", | ||
| [ |
There was a problem hiding this comment.
Consider adding a bare PSC case (example "https://p.googleapis.com").
There was a problem hiding this comment.
Additionally a case like https://[2001:db8::1]:443/mtls.googleapis.com would be good to demonstrated handling of IPv6 syntax handling
| ) | ||
|
|
||
|
|
||
| def is_mtls_endpoint(url: Optional[Union[str, bytes, Any]]) -> bool: |
There was a problem hiding this comment.
instead of Any can you use "object"? I think ideally we'd avoid usage of Any wherever possible.
| @@ -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( | |||
There was a problem hiding this comment.
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
3 participants
Fixes #18147
Follow-up to #17928
Description
This PR resolves two defects in the mTLS endpoint detection logic previously used in
requests.pyandurllib3.py:prefix in url) with proper hostname isolation viaurllib.parse.urlsplit(url).hostname. Standard non-mTLS URLs containingmtls.googleapis.comin paths or query parameters (e.g.https://storage.googleapis.com/bucket/mtls.googleapis.comorhttps://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com) will no longer trigger unnecessary certificate rotation on 401.*.p.googleapis.com) and regional mTLS domains (*.rep.mtls.googleapis.com), ensuring certificate rotation functions correctly for PSC connections._mtls_helper.is_mtls_endpoint(url)shared across bothrequestsandurllib3transports, with lazy evaluation on 401 status codes.Tests
TestIsMtlsEndpointunit test suite intests/transport/test__mtls_helper.pycovering standard mTLS, PSC endpoints, regional endpoints, path/query substring traps, port numbers, and edge cases.tests/transport/test_requests.pyandtests/transport/test_urllib3.pyverifying cert rotation is skipped on non-mTLS URLs with matching substrings and triggered on PSC URLs.