diff --git a/CHANGES.rst b/CHANGES.rst index 4966569c..16b614df 100644 --- a/CHANGES.rst +++ b/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) ^^^^^^^^^^^^^^^^^^^ diff --git a/sqlalchemy_utils/functions/database.py b/sqlalchemy_utils/functions/database.py index b0c0bc30..08930b81 100644 --- a/sqlalchemy_utils/functions/database.py +++ b/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: diff --git a/tests/functions/test_database.py b/tests/functions/test_database.py index 5f81ca50..f59f8df7 100644 --- a/tests/functions/test_database.py +++ b/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: