gh-84943: Add support for 'directonly' and 'innocuous' flags for user-defined functions

Co-authored-by: Erlend E. Aasland <erlend@python.org>

Author: jbosboom

Doc/library/sqlite3.rst

@@ -699,7 +699,7 @@ Connection objects
       :meth:`~Cursor.executescript` on it with the given *sql_script*.
       Return the new cursor object.
 
-   .. method:: create_function(name, narg, func, /, *, deterministic=False)
+   .. method:: create_function(name, narg, func, /, *, deterministic=False, innocuous=False, directonly=False)
 
       Create or remove a user-defined SQL function.
 
@@ -722,12 +722,31 @@ Connection objects
           `deterministic <https://sqlite.org/deterministic.html>`_,
           which allows SQLite to perform additional optimizations.
 
+      :param bool innocuous:
+          If ``True``, the created SQL function is marked as
+          `innocuous <https://sqlite.org/c3ref/c_deterministic.html#sqliteinnocuous>`__,
+          making it usable in views, triggers and schema structures even when
+          the ``trusted_schema`` pragma is disabled.
+
+      :param bool directonly:
+          If ``True``, the created SQL function is marked as
+          `directonly <https://sqlite.org/c3ref/c_deterministic.html#sqlitedirectonly>`__,
+          restricting its use to top-level SQL statements regardless of the
+          value of the ``trusted_schema`` pragma.
+
+      :raises NotSupportedError:
+          If called with *innocuous* or *directonly* equal to True on a version
+          of SQLite older than 3.31.0.
+
       .. versionchanged:: 3.8
          Added the *deterministic* parameter.
 
       .. versionchanged:: 3.15
          The first three parameters are now positional-only.
 
+      .. versionchanged:: next
+         Added the *innocuous* and *directonly* parameters.
+
       Example:
 
       .. doctest::
@@ -743,7 +762,7 @@ Connection objects
          >>> con.close()
 
 
-   .. method:: create_aggregate(name, n_arg, aggregate_class, /)
+   .. method:: create_aggregate(name, n_arg, aggregate_class, /, *, deterministic=False, innocuous=False, directonly=False)
 
       Create or remove a user-defined SQL aggregate function.
 
@@ -767,9 +786,33 @@ Connection objects
           Set to ``None`` to remove an existing SQL aggregate function.
       :type aggregate_class: :term:`class` | None
 
+      :param bool deterministic:
+          If ``True``, the created SQL function is marked as
+          `deterministic <https://sqlite.org/deterministic.html>`__,
+          which allows SQLite to perform additional optimizations.
+
+      :param bool innocuous:
+          If ``True``, the created SQL function is marked as
+          `innocuous <https://sqlite.org/c3ref/c_deterministic.html#sqliteinnocuous>`__,
+          making it usable in views, triggers and schema structures even when
+          the ``trusted_schema`` pragma is disabled.
+
+      :param bool directonly:
+          If ``True``, the created SQL function is marked as
+          `directonly <https://sqlite.org/c3ref/c_deterministic.html#sqlitedirectonly>`__,
+          restricting its use to top-level SQL statements regardless of the
+          value of the ``trusted_schema`` pragma.
+
+      :raises NotSupportedError:
+          If called with *innocuous* or *directonly* equal to True on a version
+          of SQLite older than 3.31.0.
+
       .. versionchanged:: 3.15
          All three parameters are now positional-only.
 
+      .. versionchanged:: next
+         Added the *deterministic*, *innocuous* and *directonly* parameters.
+
       Example:
 
       .. testcode::
@@ -800,7 +843,7 @@ Connection objects
          3
 
 
-   .. method:: create_window_function(name, num_params, aggregate_class, /)
+   .. method:: create_window_function(name, num_params, aggregate_class, /, *, deterministic=False, innocuous=False, directonly=False)
 
       Create or remove a user-defined aggregate window function.
 
@@ -825,14 +868,38 @@ Connection objects
 
           Set to ``None`` to remove an existing SQL aggregate window function.
 
+      :param bool deterministic:
+          If ``True``, the created SQL function is marked as
+          `deterministic <https://sqlite.org/deterministic.html>`__,
+          which allows SQLite to perform additional optimizations.
+
+      :param bool innocuous:
+          If ``True``, the created SQL function is marked as
+          `innocuous <https://sqlite.org/c3ref/c_deterministic.html#sqliteinnocuous>`__,
+          making it usable in views, triggers and schema structures even when
+          the ``trusted_schema`` pragma is disabled.
+
+      :param bool directonly:
+          If ``True``, the created SQL function is marked as
+          `directonly <https://sqlite.org/c3ref/c_deterministic.html#sqlitedirectonly>`__,
+          restricting its use to top-level SQL statements regardless of the
+          value of the ``trusted_schema`` pragma.
+
       :raises NotSupportedError:
           If used with a version of SQLite older than 3.25.0,
           which does not support aggregate window functions.
 
+      :raises NotSupportedError:
+          If called with *innocuous* or *directonly* equal to True on a version
+          of SQLite older than 3.31.0.
+
       :type aggregate_class: :term:`class` | None
 
       .. versionadded:: 3.11
 
+      .. versionchanged:: next
+         Added the *deterministic*, *innocuous* and *directonly* parameters.
+
       Example:
 
       .. testcode::

Lib/test/test_sqlite3/test_userfunctions.py

@@ -363,6 +363,40 @@ def test_func_deterministic_keyword_only(self):
         with self.assertRaises(TypeError):
             self.con.create_function("deterministic", 0, int, True)
 
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or higher")
+    def test_func_non_innocuous_in_trusted_env(self):
+        mock = Mock(return_value=None)
+        self.con.create_function("noninnocuous", 0, mock, innocuous=False)
+        self.con.execute("pragma trusted_schema = 0")
+        self.con.execute("create view notallowed as select noninnocuous() = noninnocuous()")
+        with self.assertRaises(sqlite.OperationalError) as cm:
+            self.con.execute("select * from notallowed")
+        self.assertEqual(str(cm.exception), 'unsafe use of noninnocuous()')
+
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or higher")
+    def test_func_innocuous_in_trusted_env(self):
+        mock = Mock(return_value=None)
+        self.con.create_function("innocuous", 0, mock, innocuous=True)
+        self.con.execute("pragma trusted_schema = 0")
+        self.con.execute("create view allowed as select innocuous() = innocuous()")
+        self.con.execute("select * from allowed")
+        self.assertEqual(mock.call_count, 2)
+
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or higher")
+    def test_func_direct_only(self):
+        mock = Mock(return_value=None)
+        self.con.create_function("directonly", 0, mock, directonly=True)
+        self.con.execute("pragma trusted_schema = 1")
+        self.con.execute("select directonly() = directonly()")
+        self.assertEqual(mock.call_count, 2)
+        self.con.execute("create view notallowed as select directonly() = directonly()")
+        with self.assertRaises(sqlite.OperationalError) as cm:
+            self.con.execute("select * from notallowed")
+        self.assertEqual(str(cm.exception), 'unsafe use of directonly()')
+
     def test_function_destructor_via_gc(self):
         # See bpo-44304: The destructor of the user function can
         # crash if is called without the GIL from the gc functions
@@ -479,6 +513,9 @@ def setUp(self):
             from test order by x
         """
         self.con.create_window_function("sumint", 1, WindowSumInt)
+        if sqlite.sqlite_version_info >= (3, 31, 0):
+            self.con.create_window_function("sumintInnocuous", 1, WindowSumInt, innocuous=True)
+            self.con.create_window_function("sumintDirectOnly", 1, WindowSumInt, directonly=True)
 
     def tearDown(self):
         self.cur.close()
@@ -488,6 +525,34 @@ def test_win_sum_int(self):
         self.cur.execute(self.query % "sumint")
         self.assertEqual(self.cur.fetchall(), self.expected)
 
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                "Requires SQLite 3.31.0 or newer")
+    def test_win_non_innocuous(self):
+        self.cur.execute("pragma trusted_schema = 0")
+        self.cur.execute("create view notallowed as " + self.query % "sumint")
+        with self.assertRaises(sqlite.OperationalError) as cm:
+            self.cur.execute("select * from notallowed")
+        self.assertEqual(str(cm.exception), 'unsafe use of sumint()')
+
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or newer")
+    def test_win_innocuous(self):
+        self.cur.execute("pragma trusted_schema = 0")
+        self.cur.execute("create view allowed as " + self.query % "sumintInnocuous")
+        self.cur.execute("select * from allowed")
+        self.assertEqual(self.cur.fetchall(), self.expected)
+
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or newer")
+    def test_win_directonly(self):
+        self.cur.execute("pragma trusted_schema = 1")
+        self.cur.execute("create view notallowed as " + self.query % "sumintDirectOnly")
+        with self.assertRaises(sqlite.OperationalError) as cm:
+            self.cur.execute("select * from notallowed")
+        self.assertEqual(str(cm.exception), 'unsafe use of sumintDirectOnly()')
+        self.cur.execute(self.query % "sumintDirectOnly")
+        self.assertEqual(self.cur.fetchall(), self.expected)
+
     def test_win_error_on_create(self):
         with self.assertRaisesRegex(sqlite.ProgrammingError, "not -100"):
             self.con.create_window_function("shouldfail", -100, WindowSumInt)
@@ -614,6 +679,9 @@ def setUp(self):
         self.con.create_aggregate("checkTypes", -1, AggrCheckTypes)
         self.con.create_aggregate("mysum", 1, AggrSum)
         self.con.create_aggregate("aggtxt", 1, AggrText)
+        if sqlite.sqlite_version_info >= (3, 31, 0):
+            self.con.create_aggregate("mysumInnocuous", 1, AggrSum, innocuous=True)
+            self.con.create_aggregate("mysumDirectOnly", 1, AggrSum, directonly=True)
 
     def tearDown(self):
         self.con.close()
@@ -705,6 +773,45 @@ def test_aggr_check_aggr_sum(self):
         val = cur.fetchone()[0]
         self.assertEqual(val, 60)
 
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or newer")
+    def test_aggr_non_innocuous(self):
+        cur = self.con.cursor()
+        cur.execute("pragma trusted_schema = 0")
+        cur.execute("delete from test")
+        cur.execute("insert into test(i) values (?)", (10,))
+        cur.execute("create view notallowed as select mysum(i) from test")
+        with self.assertRaises(sqlite.OperationalError) as cm:
+            cur.execute("select * from notallowed")
+        self.assertEqual(str(cm.exception), 'unsafe use of mysum()')
+
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or newer")
+    def test_aggr_innocuous(self):
+        cur = self.con.cursor()
+        cur.execute("pragma trusted_schema = 0")
+        cur.execute("delete from test")
+        cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)])
+        cur.execute("create view allowed as select mysumInnocuous(i) from test")
+        cur.execute("select * from allowed")
+        val = cur.fetchone()[0]
+        self.assertEqual(val, 60)
+
+    @unittest.skipIf(sqlite.sqlite_version_info < (3, 31, 0),
+                     "Requires SQLite 3.31.0 or newer")
+    def test_aggr_directonly(self):
+        cur = self.con.cursor()
+        cur.execute("pragma trusted_schema = 1")
+        cur.execute("delete from test")
+        cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)])
+        cur.execute("create view notallowed as select mysumDirectOnly(i) from test")
+        with self.assertRaises(sqlite.OperationalError) as cm:
+            cur.execute("select * from notallowed")
+        self.assertEqual(str(cm.exception), 'unsafe use of mysumDirectOnly()')
+        cur.execute("select mysumDirectOnly(i) from test")
+        val = cur.fetchone()[0]
+        self.assertEqual(val, 60)
+
     def test_aggr_no_match(self):
         cur = self.con.execute("select mysum(i) from (select 1 as i) where i == 0")
         val = cur.fetchone()[0]

Misc/NEWS.d/next/Library/2026-05-17-11-55-52.gh-issue-89483.l6ShbK.rst

@@ -0,0 +1,2 @@
+Add support for ``SQLITE_INNOCUOUS`` and ``SQLITE_DIRECTONLY`` flags in
+:mod:`sqlite3`.