gh-143198: fix SIGSEGV in sqlite3.execute[many] with re-entrant parameter iterator

Author: picnixz

Lib/test/test_sqlite3/test_dbapi.py

@@ -22,6 +22,7 @@
 
 import contextlib
 import functools
+import operator
 import os
 import sqlite3 as sqlite
 import subprocess
@@ -32,7 +33,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 +729,27 @@ def test_database_keyword(self):
             self.assertEqual(type(cx), sqlite.Connection)
 
 
+class CxWrapper:
+    def __init__(self, cx):
+        self.cx = cx
+
+
+class EvilParamsWithIter(CxWrapper):
+
+    def __iter__(self):
+        self.cx.close()
+        return iter([(1,)])
+
+
+class EvilParamsWithNext(CxWrapper):
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        self.cx.close()
+        return (1,)
+
+
 class CursorTests(unittest.TestCase):
     def setUp(self):
         self.cx = sqlite.connect(":memory:")
@@ -742,6 +764,16 @@ def tearDown(self):
         self.cu.close()
         self.cx.close()
 
+    def do_test_connection_use_after_close(self, method_name, 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:")
+        self.addCleanup(cx.close)
+        cu = cx.cursor()
+        params = params_class(cx)
+        method = operator.methodcaller(method_name, "SELECT ?", params)
+        self.assertRaises(sqlite.ProgrammingError, method, cu)
+
     def test_execute_no_args(self):
         self.cu.execute("delete from test")
 
@@ -813,6 +845,10 @@ def test_execute_non_iterable(self):
             self.cu.execute("insert into test(id) values (?)", 42)
         self.assertEqual(str(cm.exception), 'parameters are of unsupported type')
 
+    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
+    def test_execute_use_after_close(self, params_class):
+        self.do_test_connection_use_after_close("execute", params_class)
+
     def test_execute_wrong_no_of_args1(self):
         # too many parameters
         with self.assertRaises(sqlite.ProgrammingError):
@@ -1030,6 +1066,10 @@ def test_execute_many_not_iterable(self):
         with self.assertRaises(TypeError):
             self.cu.executemany("insert into test(income) values (?)", 42)
 
+    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
+    def test_executemany_use_after_close(self, params_class):
+        self.do_test_connection_use_after_close("executemany", params_class)
+
     def test_fetch_iter(self):
         # Optional DB-API extension.
         self.cu.execute("delete from test")
@@ -1645,6 +1685,15 @@ def tearDown(self):
         self.cur.close()
         self.con.close()
 
+    def do_test_connection_use_after_close(self, method_name, 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:")
+        self.addCleanup(cx.close)
+        params = params_class(cx)
+        method = operator.methodcaller(method_name, "SELECT ?", params)
+        self.assertRaises(sqlite.ProgrammingError, method, cx)
+
     def test_script_string_sql(self):
         cur = self.cur
         cur.executescript("""
@@ -1711,6 +1760,10 @@ def test_connection_execute(self):
         result = self.con.execute("select 5").fetchone()[0]
         self.assertEqual(result, 5, "Basic test of Connection.execute")
 
+    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
+    def test_connection_execute_use_after_close(self, params_class):
+        self.do_test_connection_use_after_close("execute", params_class)
+
     def test_connection_executemany(self):
         con = self.con
         con.execute("create table test(foo)")
@@ -1719,6 +1772,10 @@ 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", (EvilParamsWithIter, EvilParamsWithNext))
+    def test_connection_executemany_use_after_close(self, params_class):
+        self.do_test_connection_use_after_close("executemany", params_class)
+
     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,4 @@
+:mod:`sqlite3`: fix crashes in :func:`~sqlite3.execute`, :func:`~sqlite3.executemany`,
+:meth:`Cursor.execute <sqlite3.Cursor.execute>`, and :meth:`Cursor.executemany
+<sqlite3.Cursor.executemany>` when iterating over the query's parameters closes
+the current sqlite3 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);