fix: escape credentials and options when rebuilding the MongoDB URI

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)}")
 

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")