gh-143198: fix SIGSEGV in sqlite3.executemany with concurrent mutations

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("""

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 <sqlite3.Connection.executemany>`
+and :meth:`Cursor.executemany <sqlite3.Cursor.executemany>` when iterating over
+the query's parameters closes the current connection. Patch by Bénédikt Tran.

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()) {