From 573d425c6fe225da46d3eb0096579d6dcb1e7c33 Mon Sep 17 00:00:00 2001 From: Mehmet Salih Yavuz Date: Mon, 7 Sep 2026 12:28:35 +0300 Subject: [PATCH] fix: escape credentials and options when rebuilding the MongoDB URI SQLAlchemy percent-decodes username, password and query values when parsing the URL. create_connect_args reassembled the MongoDB URI from those decoded values without re-escaping them, so a password containing a reserved character such as @ broke pymongo's userinfo parsing even when the user had correctly encoded it as %40. Re-escape credentials and query keys/values with quote_plus, and emit repeated query keys once per value instead of stringifying the tuple. --- .../sqlalchemy_mongodb/sqlalchemy_dialect.py | 12 ++++-- tests/test_sqlalchemy_dialect.py | 37 +++++++++++++++++++ 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py b/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py index 16f2c06..540a0f1 100644 --- a/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py +++ b/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- import logging from typing import Any, Dict, List, Optional, Tuple, Type +from urllib.parse import quote_plus from sqlalchemy import pool, types from sqlalchemy.engine import default, url @@ -267,12 +268,13 @@ def create_connect_args(self, url: url.URL) -> Tuple[List[Any], Dict[str, Any]]: # Start with scheme (mongodb only - srv handled separately) uri_parts.append(f"{url.drivername}://") - # Add credentials if present + # SQLAlchemy has already percent-decoded the credentials, so they must be + # re-escaped (RFC 3986) before pymongo parses the rebuilt URI. if url.username: if url.password: - uri_parts.append(f"{url.username}:{url.password}@") + uri_parts.append(f"{quote_plus(url.username)}:{quote_plus(url.password)}@") else: - uri_parts.append(f"{url.username}@") + uri_parts.append(f"{quote_plus(url.username)}@") # Add host and port if url.host: @@ -288,7 +290,9 @@ def create_connect_args(self, url: url.URL) -> Tuple[List[Any], Dict[str, Any]]: if url.query: query_parts = [] for key, value in url.query.items(): - query_parts.append(f"{key}={value}") + values = value if isinstance(value, (list, tuple)) else (value,) + for item in values: + query_parts.append(f"{quote_plus(key)}={quote_plus(str(item))}") if query_parts: uri_parts.append(f"?{'&'.join(query_parts)}") diff --git a/tests/test_sqlalchemy_dialect.py b/tests/test_sqlalchemy_dialect.py index e2fe7b2..fa5591b 100644 --- a/tests/test_sqlalchemy_dialect.py +++ b/tests/test_sqlalchemy_dialect.py @@ -102,6 +102,43 @@ def test_create_connect_args_with_auth(self): self.assertIn("host", kwargs) self.assertEqual(kwargs["host"], "mongodb://user:pass@localhost:27017/testdb") + def test_create_connect_args_escapes_credentials(self): + """Credentials with reserved characters must be re-escaped in the rebuilt URI.""" + test_url = url.make_url("mongodb://us%40er:p%40ss%3Aw%2Frd@localhost:27017/testdb") + args, kwargs = self.dialect.create_connect_args(test_url) + + self.assertEqual(kwargs["host"], "mongodb://us%40er:p%40ss%3Aw%2Frd@localhost:27017/testdb") + + from pymongo.uri_parser import parse_uri + + parsed = parse_uri(kwargs["host"]) + self.assertEqual(parsed["username"], "us@er") + self.assertEqual(parsed["password"], "p@ss:w/rd") + + def test_create_connect_args_escapes_username_only(self): + """Username without password is also re-escaped.""" + test_url = url.make_url("mongodb://us%40er@localhost:27017/testdb") + args, kwargs = self.dialect.create_connect_args(test_url) + + self.assertEqual(kwargs["host"], "mongodb://us%40er@localhost:27017/testdb") + + def test_create_connect_args_escapes_query_values(self): + """Query option values with reserved characters are re-escaped.""" + test_url = url.make_url("mongodb://localhost/testdb?authMechanismProperties=SERVICE_NAME%3Amongo%26x") + args, kwargs = self.dialect.create_connect_args(test_url) + + self.assertEqual(kwargs["host"], "mongodb://localhost/testdb?authMechanismProperties=SERVICE_NAME%3Amongo%26x") + + def test_create_connect_args_repeated_query_key(self): + """Repeated query keys are emitted once per value instead of stringified as a tuple.""" + test_url = url.make_url("mongodb://localhost/testdb?readPreferenceTags=dc:east&readPreferenceTags=dc:west") + args, kwargs = self.dialect.create_connect_args(test_url) + + self.assertEqual( + kwargs["host"], + "mongodb://localhost/testdb?readPreferenceTags=dc%3Aeast&readPreferenceTags=dc%3Awest", + ) + def test_create_connect_args_with_query_params(self): """Test connection args with query parameters.""" test_url = url.make_url("mongodb://localhost/testdb?ssl=true&replicaSet=rs0")