From: dxbjavid Date: Wed, 17 Jun 2026 12:49:06 +0000 (-0400) Subject: quote driver name and pass-through keys in pyodbc connect string X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=c039c9fa57e68443ba87fdff37bd26a8617320c1;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git quote driver name and pass-through keys in pyodbc connect string Tightened the construction of the ODBC connection string in the pyodbc connector (as well as the mssql-python connector in 2.1) so that the driver name, the names of pass-through connection parameters, and values containing ``}`` are brace-quoted. Previously a ``}`` in the driver name or in a pass-through value, or a ``;`` in the name of a pass-through parameter, could close the surrounding token early and allow the remainder of the string to be interpreted as additional connection attributes. Pull request courtesy dxbjavid. Fixes: #13380 Closes: #13379 Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/13379 Pull-request-sha: 34433408bff86a9044be5d431b31688ab6d7578d Change-Id: I5a0b522e62d306cf6dc59c7f61fe584df02a948d (cherry picked from commit 8604972666978ca981623c7f415e405bc074663a) --- diff --git a/doc/build/changelog/unreleased_20/13380.rst b/doc/build/changelog/unreleased_20/13380.rst new file mode 100644 index 0000000000..6d70d79df5 --- /dev/null +++ b/doc/build/changelog/unreleased_20/13380.rst @@ -0,0 +1,12 @@ +.. change:: + :tags: bug, mssql + :tickets: 13380 + + Tightened the construction of the ODBC connection string in the pyodbc + connector (as well as the mssql-python connector in 2.1) so that the + driver name, the names of pass-through connection parameters, and values + containing ``}`` are brace-quoted. Previously a ``}`` in the driver name + or in a pass-through value, or a ``;`` in the name of a pass-through + parameter, could close the surrounding token early and allow the + remainder of the string to be interpreted as additional connection + attributes. Pull request courtesy dxbjavid. diff --git a/lib/sqlalchemy/connectors/pyodbc.py b/lib/sqlalchemy/connectors/pyodbc.py index ac327b86d4..5b11694618 100644 --- a/lib/sqlalchemy/connectors/pyodbc.py +++ b/lib/sqlalchemy/connectors/pyodbc.py @@ -77,10 +77,16 @@ class PyODBCConnector(Connector): else: def check_quote(token: str) -> str: - if ";" in str(token) or str(token).startswith("{"): + if ( + ";" in str(token) + or "}" in str(token) + or str(token).startswith("{") + ): token = "{%s}" % token.replace("}", "}}") return token + driver = keys.pop("driver", self.pyodbc_driver_name) + keys = {k: check_quote(v) for k, v in keys.items()} dsn_connection = "dsn" in keys or ( @@ -96,7 +102,6 @@ class PyODBCConnector(Connector): port = ",%d" % int(keys.pop("port")) connectors = [] - driver = keys.pop("driver", self.pyodbc_driver_name) if driver is None and keys: # note if keys is empty, this is a totally blank URL util.warn( @@ -105,7 +110,9 @@ class PyODBCConnector(Connector): "DSN-less connections" ) else: - connectors.append("DRIVER={%s}" % driver) + connectors.append( + "DRIVER={%s}" % str(driver).replace("}", "}}") + ) connectors.extend( [ @@ -136,7 +143,9 @@ class PyODBCConnector(Connector): "AutoTranslate=%s" % keys.pop("odbc_autotranslate") ) - connectors.extend(["%s=%s" % (k, v) for k, v in keys.items()]) + connectors.extend( + ["%s=%s" % (check_quote(k), v) for k, v in keys.items()] + ) return ((";".join(connectors),), connect_args) diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py index a1b2e9a79c..f242ad8897 100644 --- a/test/dialect/mssql/test_engine.py +++ b/test/dialect/mssql/test_engine.py @@ -236,6 +236,11 @@ class ParseConnectTest(fixtures.TestBase): connection, ) + @testing.combinations( + ("pyodbc", pyodbc, "mssql+pyodbc"), + argnames="dialect_mod, url_prefix", + id_="iaa", + ) @testing.combinations( ( "original", @@ -245,11 +250,15 @@ class ParseConnectTest(fixtures.TestBase): "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}", - ), + "driver=foob", + { + "pyodbc": ( + "DRIVER={foob};Server=somehost%3BPORT%3D50001;" + "Database=somedb%3BPORT%3D50001;" + "UID={someuser;PORT=50001};" + "PWD={some{strange}}pw;PORT=50001}", + ), + }, ), ( "issue_8062", @@ -259,22 +268,92 @@ class ParseConnectTest(fixtures.TestBase): "localhost", "mydb", ), + "driver=foob", + { + "pyodbc": ( + "DRIVER={foob};Server=localhost;" + "Database=mydb;UID=larry;" + "PWD={{moe}", + ), + }, + ), + ( + # a driver name that starts with { and contains } looks like + # a complete brace-quoted token. per the MS-ODBCSTR spec, + # only the right brace is escaped (doubled) within a + # brace-enclosed expression; the left brace is passed + # through as a literal character. + "driver_brace_open_injection", ( - "DRIVER={foob};Server=localhost;" - "Database=mydb;UID=larry;" - "PWD={{moe}", + "username", + "password", + "hostspec", + "database", ), + "driver=%7BUID%3Devil%7D", + { + "pyodbc": ( + "DRIVER={{UID=evil}}};" + "Server=hostspec;Database=database;" + "UID=username;PWD=password", + ), + }, ), - argnames="tokens, connection_string", - id_="iaa", + ( + # the driver name is always wrapped in braces in pyodbc, so a + # closing brace in the value has to be escaped, otherwise it + # ends the brace early and the remainder is read as further + # attributes + "driver_brace_injection", + ( + "username", + "password", + "hostspec", + "database", + ), + "driver=foob%7DUID%3Devil", + { + "pyodbc": ( + "DRIVER={foob}}UID=evil};" + "Server=hostspec;Database=database;" + "UID=username;PWD=password", + ), + }, + ), + ( + # names of pass-through parameters get the same brace quoting + # as their values, so a semicolon in a parameter name can't be + # used to smuggle in an extra attribute + "pass_through_key_injection", + ( + "username", + "password", + "hostspec", + "database", + ), + "driver=foob&foo%3BUID=evil", + { + "pyodbc": ( + "DRIVER={foob};Server=hostspec;Database=database;" + "UID=username;PWD=password;{foo;UID}=evil", + ), + }, + ), + argnames="tokens, query, expected", + id_="iaaa", ) - 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() + def test_token_injection( + self, dialect_mod, url_prefix, tokens, query, expected + ): + dialect_key = dialect_mod.__name__.rsplit(".", 1)[-1] + u = url.make_url( + "%s://%s:%s@%s/%s?%s" % ((url_prefix,) + tokens + (query,)) + ) + dialect = dialect_mod.dialect() connection = dialect.create_connect_args(u) eq_( ( - connection_string, + expected[dialect_key], {}, ), connection,