mssql login failure if password starts with "{"

Fix issue where a password with a leading "{" would
result in login failure.

Fixes: #8062
Change-Id: If91c2c211937b5eac89b8d525c22a19b0a94c5c4

3 files changed, 41 insertions, 16 deletions

diff --git a/doc/build/changelog/unreleased_14/8062.rst b/doc/build/changelog/unreleased_14/8062.rst
new file mode 100644
index 000000000..ada473de9
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8062.rst
@@ -0,0 +1,5 @@
+.. change::
+    :tags: bug, mssql
+    :tickets: 8062
+
+    Fix issue where a password with a leading "{" would result in login failure.
diff --git a/lib/sqlalchemy/connectors/pyodbc.py b/lib/sqlalchemy/connectors/pyodbc.py
index c24fa344b..e24acda75 100644
--- a/lib/sqlalchemy/connectors/pyodbc.py
+++ b/lib/sqlalchemy/connectors/pyodbc.py
@@ -77,7 +77,7 @@ class PyODBCConnector(Connector):
         else:
 
             def check_quote(token: str) -> str:
-                if ";" in str(token):
+                if ";" in str(token) or str(token).startswith("{"):
                     token = "{%s}" % token.replace("}", "}}")
                 return token
 
diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py
index d54a37ceb..296820539 100644
--- a/test/dialect/mssql/test_engine.py
+++ b/test/dialect/mssql/test_engine.py
@@ -235,25 +235,45 @@ class ParseConnectTest(fixtures.TestBase):
             connection,
         )
 
-    def test_pyodbc_token_injection(self):
-        token1 = "someuser%3BPORT%3D50001"
-        token2 = "some{strange}pw%3BPORT%3D50001"
-        token3 = "somehost%3BPORT%3D50001"
-        token4 = "somedb%3BPORT%3D50001"
-
-        u = url.make_url(
-            "mssql+pyodbc://%s:%s@%s/%s?driver=foob"
-            % (token1, token2, token3, token4)
-        )
+    @testing.combinations(
+        (
+            "original",
+            (
+                "someuser%3BPORT%3D50001",
+                "some{strange}pw%3BPORT%3D50001",
+                "somehost%3BPORT%3D50001",
+                "somedb%3BPORT%3D50001",
+            ),
+            (
+                "DRIVER={foob};Server=somehost%3BPORT%3D50001;"
+                "Database=somedb%3BPORT%3D50001;UID={someuser;PORT=50001};"
+                "PWD={some{strange}}pw;PORT=50001}",
+            ),
+        ),
+        (
+            "issue_8062",
+            (
+                "larry",
+                "{moe",
+                "localhost",
+                "mydb",
+            ),
+            (
+                "DRIVER={foob};Server=localhost;"
+                "Database=mydb;UID=larry;"
+                "PWD={{moe}",
+            ),
+        ),
+        argnames="tokens, connection_string",
+        id_="iaa",
+    )
+    def test_pyodbc_token_injection(self, tokens, connection_string):
+        u = url.make_url("mssql+pyodbc://%s:%s@%s/%s?driver=foob" % tokens)
         dialect = pyodbc.dialect()
         connection = dialect.create_connect_args(u)
         eq_(
             (
-                (
-                    "DRIVER={foob};Server=somehost%3BPORT%3D50001;"
-                    "Database=somedb%3BPORT%3D50001;UID={someuser;PORT=50001};"
-                    "PWD={some{strange}}pw;PORT=50001}",
-                ),
+                connection_string,
                 {},
             ),
             connection,