Movatterモバイル変換


[0]ホーム

URL:


Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

Sign up
Appearance settings

[3.14] gh-142830: prevent some crashes when mutatingsqlite3 callbacks (GH-143245)#143322

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Merged
picnixz merged 1 commit intopython:3.14frompicnixz:backport-7f6c16a-3.14
Jan 1, 2026
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 119 additions & 2 deletionsLib/test/test_sqlite3/test_hooks.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,11 +24,15 @@
import sqlite3 as sqlite
import unittest

from test.support import import_helper
from test.support.os_helper import TESTFN, unlink

from .util import memory_database, cx_limit, with_tracebacks
from .util import MemoryDatabaseMixin

# TODO(picnixz): increase test coverage for other callbacks
# such as 'func', 'step', 'finalize', and 'collation'.


class CollationTests(MemoryDatabaseMixin, unittest.TestCase):

Expand DownExpand Up@@ -129,8 +133,55 @@ def test_deregister_collation(self):
self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll')


class AuthorizerTests(MemoryDatabaseMixin, unittest.TestCase):

def assert_not_authorized(self, func, /, *args, **kwargs):
with self.assertRaisesRegex(sqlite.DatabaseError, "not authorized"):
func(*args, **kwargs)

# When a handler has an invalid signature, the exception raised is
# the same that would be raised if the handler "negatively" replied.

def test_authorizer_invalid_signature(self):
self.cx.execute("create table if not exists test(a number)")
self.cx.set_authorizer(lambda: None)
self.assert_not_authorized(self.cx.execute, "select * from test")

# Tests for checking that callback context mutations do not crash.
# Regression tests for https://github.com/python/cpython/issues/142830.

@with_tracebacks(ZeroDivisionError, regex="hello world")
def test_authorizer_concurrent_mutation_in_call(self):
self.cx.execute("create table if not exists test(a number)")

def handler(*a, **kw):
self.cx.set_authorizer(None)
raise ZeroDivisionError("hello world")

self.cx.set_authorizer(handler)
self.assert_not_authorized(self.cx.execute, "select * from test")

@with_tracebacks(OverflowError)
def test_authorizer_concurrent_mutation_with_overflown_value(self):
_testcapi = import_helper.import_module("_testcapi")
self.cx.execute("create table if not exists test(a number)")

def handler(*a, **kw):
self.cx.set_authorizer(None)
# We expect 'int' at the C level, so this one will raise
# when converting via PyLong_Int().
return _testcapi.INT_MAX + 1

self.cx.set_authorizer(handler)
self.assert_not_authorized(self.cx.execute, "select * from test")


class ProgressTests(MemoryDatabaseMixin, unittest.TestCase):

def assert_interrupted(self, func, /, *args, **kwargs):
with self.assertRaisesRegex(sqlite.OperationalError, "interrupted"):
func(*args, **kwargs)

def test_progress_handler_used(self):
"""
Test that the progress handler is invoked once it is set.
Expand DownExpand Up@@ -219,7 +270,7 @@ def bad_progress():
create table foo(a, b)
""")

deftest_progress_handler_keyword_args(self):
deftest_set_progress_handler_keyword_args(self):
regex = (
r"Passing keyword argument 'progress_handler' to "
r"_sqlite3.Connection.set_progress_handler\(\) is deprecated. "
Expand All@@ -231,6 +282,43 @@ def test_progress_handler_keyword_args(self):
self.con.set_progress_handler(progress_handler=lambda: None, n=1)
self.assertEqual(cm.filename, __file__)

# When a handler has an invalid signature, the exception raised is
# the same that would be raised if the handler "negatively" replied.

def test_progress_handler_invalid_signature(self):
self.cx.execute("create table if not exists test(a number)")
self.cx.set_progress_handler(lambda x: None, 1)
self.assert_interrupted(self.cx.execute, "select * from test")

# Tests for checking that callback context mutations do not crash.
# Regression tests for https://github.com/python/cpython/issues/142830.

@with_tracebacks(ZeroDivisionError, regex="hello world")
def test_progress_handler_concurrent_mutation_in_call(self):
self.cx.execute("create table if not exists test(a number)")

def handler(*a, **kw):
self.cx.set_progress_handler(None, 1)
raise ZeroDivisionError("hello world")

self.cx.set_progress_handler(handler, 1)
self.assert_interrupted(self.cx.execute, "select * from test")

def test_progress_handler_concurrent_mutation_in_conversion(self):
self.cx.execute("create table if not exists test(a number)")

class Handler:
def __bool__(_):
# clear the progress handler
self.cx.set_progress_handler(None, 1)
raise ValueError # force PyObject_True() to fail

self.cx.set_progress_handler(Handler.__init__, 1)
self.assert_interrupted(self.cx.execute, "select * from test")

# Running with tracebacks makes the second execution of this
# function raise another exception because of a database change.


class TraceCallbackTests(MemoryDatabaseMixin, unittest.TestCase):

Expand DownExpand Up@@ -352,7 +440,7 @@ def test_trace_bad_handler(self):
cx.set_trace_callback(lambda stmt: 5/0)
cx.execute("select 1")

deftest_trace_keyword_args(self):
deftest_set_trace_callback_keyword_args(self):
regex = (
r"Passing keyword argument 'trace_callback' to "
r"_sqlite3.Connection.set_trace_callback\(\) is deprecated. "
Expand All@@ -364,6 +452,35 @@ def test_trace_keyword_args(self):
self.con.set_trace_callback(trace_callback=lambda: None)
self.assertEqual(cm.filename, __file__)

# When a handler has an invalid signature, the exception raised is
# the same that would be raised if the handler "negatively" replied,
# but for the trace handler, exceptions are never re-raised (only
# printed when needed).

@with_tracebacks(
TypeError,
regex=r".*<lambda>\(\) missing 6 required positional arguments",
)
def test_trace_handler_invalid_signature(self):
self.cx.execute("create table if not exists test(a number)")
self.cx.set_trace_callback(lambda x, y, z, t, a, b, c: None)
self.cx.execute("select * from test")

# Tests for checking that callback context mutations do not crash.
# Regression tests for https://github.com/python/cpython/issues/142830.

@with_tracebacks(ZeroDivisionError, regex="hello world")
def test_trace_callback_concurrent_mutation_in_call(self):
self.cx.execute("create table if not exists test(a number)")

def handler(statement):
# clear the progress handler
self.cx.set_trace_callback(None)
raise ZeroDivisionError("hello world")

self.cx.set_trace_callback(handler)
self.cx.execute("select * from test")


if __name__ == "__main__":
unittest.main()
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
:mod:`sqlite3`: fix use-after-free crashes when the connection's callbacks
are mutated during a callback execution. Patch by Bénédikt Tran.
Loading
Loading

[8]ページ先頭

©2009-2026 Movatter.jp