Fix SQL injection in database_exists()

CHANGES.rst

@@ -12,6 +12,7 @@ Unreleased
 - Fix ``ChoiceType`` returning the raw scalar instead of a ``Choice`` for falsy codes such as ``0`` or the empty string. (#813)
 
   NULL values continue to return ``None``.
+- Fix SQL injection in ``database_exists()`` via the database name for postgresql, mysql and mssql. (#760)
 
 0.42.1 (2025-12-12)
 ^^^^^^^^^^^^^^^^^^^

sqlalchemy_utils/functions/database.py

@@ -421,9 +421,9 @@ def _set_url_database(url: sa.engine.url.URL, database):
     return ret
 
 
-def _get_scalar_result(engine, sql):
+def _get_scalar_result(engine, sql, params=None):
     with engine.connect() as conn:
-        return conn.scalar(sql)
+        return conn.scalar(sql, params or {})
 
 
 def _sqlite_file_exists(database):
@@ -463,12 +463,16 @@ def database_exists(url):
     engine = None
     try:
         if dialect_name == 'postgresql':
-            text = "SELECT 1 FROM pg_database WHERE datname='%s'" % database
+            text = 'SELECT 1 FROM pg_database WHERE datname=:database'
             for db in (database, 'postgres', 'template1', 'template0', None):
                 url = _set_url_database(url, database=db)
                 engine = sa.create_engine(url, isolation_level='AUTOCOMMIT')
                 try:
-                    return bool(_get_scalar_result(engine, sa.text(text)))
+                    return bool(
+                        _get_scalar_result(
+                            engine, sa.text(text), {'database': database}
+                        )
+                    )
                 except (ProgrammingError, OperationalError):
                     pass
             return False
@@ -478,9 +482,11 @@ def database_exists(url):
             engine = sa.create_engine(url)
             text = (
                 'SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA '
-                "WHERE SCHEMA_NAME = '%s'" % database
+                'WHERE SCHEMA_NAME = :database'
+            )
+            return bool(
+                _get_scalar_result(engine, sa.text(text), {'database': database})
             )
-            return bool(_get_scalar_result(engine, sa.text(text)))
 
         elif dialect_name == 'sqlite':
             url = _set_url_database(url, database=None)
@@ -492,11 +498,13 @@ def database_exists(url):
                 # not required, thus we should support that use case.
                 return True
         elif dialect_name == 'mssql':
-            text = "SELECT 1 FROM sys.databases WHERE name = '%s'" % database
+            text = 'SELECT 1 FROM sys.databases WHERE name = :database'
             url = _set_url_database(url, database='master')
             engine = sa.create_engine(url, isolation_level='AUTOCOMMIT')
             try:
-                return bool(_get_scalar_result(engine, sa.text(text)))
+                return bool(
+                    _get_scalar_result(engine, sa.text(text), {'database': database})
+                )
             except (ProgrammingError, OperationalError):
                 return False
         else:

tests/functions/test_database.py

@@ -19,6 +19,57 @@ def test_create_and_drop(self, dsn):
         assert not database_exists(dsn)
 
 
+@pytest.mark.parametrize('dialect', ['postgresql', 'mysql', 'mssql'])
+def test_database_exists_binds_database_name_as_parameter(
+    dialect, monkeypatch
+):
+    """Regression test for GH-760.
+
+    ``database_exists`` used to build its existence-check query with
+    ``%``-string interpolation, so a database name containing SQL
+    metacharacters could inject additional statements. The database name
+    must instead be sent as a bound parameter, never spliced into the SQL
+    text.
+    """
+    payload = "x'; CREATE TABLE hacked (id int); --"
+    url = f'{dialect}://user:pass@localhost/{payload}'
+
+    captured = []
+
+    def record(conn, cursor, statement, parameters, context, executemany):
+        captured.append((statement, parameters))
+
+    real_create_engine = sa.create_engine
+
+    # Redirect every engine created by database_exists() to an in-memory
+    # sqlite database so we can inspect exactly what gets sent to the
+    # DBAPI, without needing a live postgres/mysql/mssql server. The
+    # query itself will fail against sqlite (wrong system tables), which
+    # is fine -- database_exists() treats that as "does not exist".
+    def fake_create_engine(*args, **kwargs):
+        engine = real_create_engine('sqlite://')
+        sa.event.listen(engine, 'before_cursor_execute', record)
+        return engine
+
+    monkeypatch.setattr(sa, 'create_engine', fake_create_engine)
+
+    # The stub sqlite engine lacks the real system tables/views that
+    # postgres/mysql/mssql would query, so the statement itself fails.
+    # database_exists() only swallows that failure for postgres/mssql;
+    # what we actually care about here is what was sent to the DBAPI.
+    try:
+        database_exists(url)
+    except (sa.exc.OperationalError, sa.exc.ProgrammingError):
+        pass
+
+    assert captured, 'expected database_exists() to execute a query'
+    for statement, parameters in captured:
+        # The malicious payload must never appear inside the SQL text
+        # itself -- it may only appear as a bound parameter value.
+        assert payload not in statement
+        assert 'hacked' not in statement
+
+
 @pytest.mark.usefixtures('sqlite_memory_dsn')
 class TestDatabaseSQLiteMemory: