From 36ae843bded62385ba84cbd4f29d07ce76712e4f Mon Sep 17 00:00:00 2001 From: Qi Deng Date: Wed, 29 Apr 2026 13:55:08 -0700 Subject: [PATCH] URL-encode client_id in Azure IMDS token request The `_get_azure_response()` function constructs the Azure IMDS URL by interpolating `client_id` via f-string without URL encoding. While `resource` is already encoded (via `quote()` at the call site in `auth_oidc_shared.py`), `client_id` is not, creating an inconsistency. Apply `urllib.parse.quote()` to `client_id` before interpolation, consistent with the handling of `resource` and with the Node.js driver's use of `url.searchParams.append()` for the same parameter. Add a test to verify special characters in `client_id` are properly percent-encoded and cannot introduce additional query parameters. --- pymongo/_azure_helpers.py | 3 ++- test/test_azure_helpers.py | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/pymongo/_azure_helpers.py b/pymongo/_azure_helpers.py index 8a7af0b407..b7a5181857 100644 --- a/pymongo/_azure_helpers.py +++ b/pymongo/_azure_helpers.py @@ -17,6 +17,7 @@ import json from typing import Any, Optional +from urllib.parse import quote def _get_azure_response( @@ -29,7 +30,7 @@ def _get_azure_response( url += "?api-version=2018-02-01" url += f"&resource={resource}" if client_id: - url += f"&client_id={client_id}" + url += f"&client_id={quote(client_id)}" headers = {"Metadata": "true", "Accept": "application/json"} request = Request(url, headers=headers) # noqa: S310 try: diff --git a/test/test_azure_helpers.py b/test/test_azure_helpers.py index 6fe6451877..25d1a779c6 100644 --- a/test/test_azure_helpers.py +++ b/test/test_azure_helpers.py @@ -150,6 +150,20 @@ def test_timeout_passed_to_urlopen(self): _, kwargs = mock_open.call_args self.assertEqual(kwargs["timeout"], 42) + def test_client_id_is_url_encoded(self): + """Ensure special characters in client_id are percent-encoded.""" + body = json.dumps({"access_token": "tok", "expires_in": "3600"}) + with _mock_urlopen(200, body) as mock_open: + self._call(client_id="id with spaces&special=chars") + + url = mock_open.call_args[0][0].full_url + # '&' and '=' must be percent-encoded so they don't inject extra query params + self.assertIn("client_id=id%20with%20spaces%26special%3Dchars", url) + # The encoded client_id should not introduce a raw '&' + # Count params: api-version, resource, client_id — exactly 3 + query_string = url.split("?", 1)[1] + self.assertEqual(query_string.count("&"), 2) + if __name__ == "__main__": unittest.main()