diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py index 20e39f61e4dedb..323a776c1e2cdf 100644 --- a/Lib/test/test_sqlite3/test_dbapi.py +++ b/Lib/test/test_sqlite3/test_dbapi.py @@ -32,7 +32,7 @@ import warnings from test.support import ( - SHORT_TIMEOUT, check_disallow_instantiation, requires_subprocess + SHORT_TIMEOUT, check_disallow_instantiation, requires_subprocess, subTests ) from test.support import gc_collect from test.support import threading_helper, import_helper @@ -728,6 +728,33 @@ def test_database_keyword(self): self.assertEqual(type(cx), sqlite.Connection) +class CxWrapper: + def __init__(self, cx): + self.cx = cx + + def side_effect(self): + self.cx.close() + + +class ParamsCxCloseInIterMany(CxWrapper): + def __iter__(self): + self.side_effect() + return iter([(1,), (2,), (3,)]) + + +class ParamsCxCloseInNext(CxWrapper): + def __init__(self, cx): + super().__init__(cx) + self.r = iter(range(10)) + + def __iter__(self): + return self + + def __next__(self): + self.side_effect() + return (next(self.r),) + + class CursorTests(unittest.TestCase): def setUp(self): self.cx = sqlite.connect(":memory:") @@ -1030,6 +1057,18 @@ def test_execute_many_not_iterable(self): with self.assertRaises(TypeError): self.cu.executemany("insert into test(income) values (?)", 42) + @subTests("params_class", (ParamsCxCloseInIterMany, ParamsCxCloseInNext)) + def test_executemany_use_after_close(self, params_class): + # Prevent SIGSEGV with iterable of parameters closing the connection. + # Regression test for https://github.com/python/cpython/issues/143198. + cx = sqlite.connect(":memory:") + cx.execute("create table tmp(a number)") + self.addCleanup(cx.close) + cu = cx.cursor() + msg = r"Cannot operate on a closed database\." + with self.assertRaisesRegex(sqlite.ProgrammingError, msg): + cu.executemany("insert into tmp(a) values (?)", params_class(cx)) + def test_fetch_iter(self): # Optional DB-API extension. self.cu.execute("delete from test") @@ -1719,6 +1758,17 @@ def test_connection_executemany(self): self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany") self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany") + @subTests("params_class", (ParamsCxCloseInIterMany, ParamsCxCloseInNext)) + def test_connection_executemany_use_after_close(self, params_class): + # Prevent SIGSEGV with iterable of parameters closing the connection. + # Regression test for https://github.com/python/cpython/issues/143198. + cx = sqlite.connect(":memory:") + cx.execute("create table tmp(a number)") + self.addCleanup(cx.close) + msg = r"Cannot operate on a closed database\." + with self.assertRaisesRegex(sqlite.ProgrammingError, msg): + cx.executemany("insert into tmp(a) values (?)", params_class(cx)) + def test_connection_executescript(self): con = self.con con.executescript(""" diff --git a/Misc/NEWS.d/next/Library/2025-12-27-10-36-18.gh-issue-143198.DdIHyC.rst b/Misc/NEWS.d/next/Library/2025-12-27-10-36-18.gh-issue-143198.DdIHyC.rst new file mode 100644 index 00000000000000..dd433dbd22104c --- /dev/null +++ b/Misc/NEWS.d/next/Library/2025-12-27-10-36-18.gh-issue-143198.DdIHyC.rst @@ -0,0 +1,3 @@ +:mod:`sqlite3`: fix crashes in :meth:`Connection.executemany ` +and :meth:`Cursor.executemany ` when iterating over +the query's parameters closes the current connection. Patch by Bénédikt Tran. diff --git a/Modules/_sqlite/cursor.c b/Modules/_sqlite/cursor.c index 4611c9e5e3e437..91968d2a3d2569 100644 --- a/Modules/_sqlite/cursor.c +++ b/Modules/_sqlite/cursor.c @@ -870,6 +870,12 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation } } + // PyObject_GetIter() may have a side-effect on the connection's state. + // See: https://github.com/python/cpython/issues/143198. + if (!pysqlite_check_connection(self->connection)) { + goto error; + } + /* reset description */ Py_INCREF(Py_None); Py_SETREF(self->description, Py_None); @@ -925,6 +931,11 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation if (!parameters) { break; } + // PyIter_Next() may have a side-effect on the connection's state. + // See: https://github.com/python/cpython/issues/143198. + if (!pysqlite_check_connection(self->connection)) { + goto error; + } bind_parameters(state, self->statement, parameters); if (PyErr_Occurred()) {