From 9f66cf72c1c9170f10e863d750ac4eee47113a7f Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Tue, 26 May 2026 21:42:50 -0700
Subject: [PATCH 1/8] Removed execute write SQL from query create page

---
 datasette/templates/query_create.html | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/datasette/templates/query_create.html b/datasette/templates/query_create.html
index f5dadbff..ec910456 100644
--- a/datasette/templates/query_create.html
+++ b/datasette/templates/query_create.html
@@ -106,9 +106,6 @@ form.sql .query-create-sql textarea#sql-editor {
 .query-create-analysis-note {
     margin: 0;
 }
-.query-create-action {
-    margin: 0.35rem 0 1rem;
-}
 .query-create-analysis {
     margin-top: 0.8rem;
 }
@@ -171,10 +168,6 @@ form.sql .query-create-sql textarea#sql-editor {
         <label><input type="checkbox" name="is_private" value="1"{% if is_private %} checked{% endif %}> Private</label>
         <span class="query-create-option-note">Queries marked private can only be seen by you, their creator.</span>
     </p>
-    {% if sql and analysis_is_write %}
-        <p class="query-create-action"><a href="{{ urls.database(database) }}/-/execute-write?{{ {'sql': sql}|urlencode|safe }}">Execute write SQL</a></p>
-    {% endif %}
-
     <p class="query-create-submit"><input type="submit" value="Save query" data-query-create-submit{% if save_disabled %} disabled{% endif %}></p>
 
     <div class="query-create-analysis" id="query-create-analysis-section"{% if not has_sql %} hidden{% endif %}>

From 737ff03efbb2bdc99b10d2654b7818526ec51e13 Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Tue, 26 May 2026 22:11:06 -0700
Subject: [PATCH 2/8] Expanded analysis of SQL operations, refs #2748

---
 datasette/permissions.py         |  10 ++
 datasette/stored_queries.py      | 137 +++++++++++++--
 datasette/utils/sql_analysis.py  | 289 +++++++++++++++++++++++++++----
 datasette/views/execute_write.py |   9 +-
 datasette/views/query_helpers.py | 104 +++++++----
 tests/test_actions_sql.py        |  14 +-
 tests/test_internals_database.py |  34 ++--
 tests/test_queries.py            | 166 ++++++++++++++++++
 tests/test_utils_sql_analysis.py |  97 +++++++++--
 9 files changed, 740 insertions(+), 120 deletions(-)

diff --git a/datasette/permissions.py b/datasette/permissions.py
index 917c58ab..a9a3cc7c 100644
--- a/datasette/permissions.py
+++ b/datasette/permissions.py
@@ -58,6 +58,16 @@ class Resource(ABC):
         self.child = child
         self._private = None  # Sentinel to track if private was set
 
+    def __str__(self) -> str:
+        return "/".join(
+            str(part) for part in (self.parent, self.child) if part is not None
+        )
+
+    def __repr__(self) -> str:
+        return "{}(parent={!r}, child={!r})".format(
+            self.__class__.__name__, self.parent, self.child
+        )
+
     @property
     def private(self) -> bool:
         """
diff --git a/datasette/stored_queries.py b/datasette/stored_queries.py
index bcfdfdb4..c4b083e5 100644
--- a/datasette/stored_queries.py
+++ b/datasette/stored_queries.py
@@ -2,11 +2,16 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 import json
-from typing import Any, Iterable
+from typing import Any, Iterable, TYPE_CHECKING
 
-from .resources import TableResource
+from .resources import DatabaseResource, TableResource
+from .permissions import Resource
 from .utils import named_parameters, sqlite3, tilde_encode, urlsafe_components
 from .utils.asgi import Forbidden
+from .utils.sql_analysis import Operation, SQLAnalysis
+
+if TYPE_CHECKING:
+    from .app import Datasette
 
 UNCHANGED = object()
 
@@ -583,20 +588,94 @@ async def list_queries(
     )
 
 
-async def ensure_query_write_permissions(
-    datasette: Any,
-    database: str,
-    sql: str,
-    *,
-    actor: dict[str, Any] | None = None,
-    params: dict[str, Any] | None = None,
-    analysis: Any = None,
-) -> Any:
+PermissionRequirement = tuple[str, Resource]
+
+
+def permission_for_operation(operation: Operation) -> PermissionRequirement | None:
     write_actions = {
         "insert": "insert-row",
         "update": "update-row",
         "delete": "delete-row",
     }
+    action = write_actions.get(operation.operation)
+    if (
+        action
+        and operation.target_type == "table"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return (
+            action,
+            TableResource(database=operation.database, table=operation.table),
+        )
+    if operation.operation == "create" and operation.target_type == "table":
+        if operation.database is None:
+            return None
+        return (
+            "create-table",
+            DatabaseResource(database=operation.database),
+        )
+    if (
+        operation.operation == "alter"
+        and operation.target_type == "table"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return (
+            "alter-table",
+            TableResource(database=operation.database, table=operation.table),
+        )
+    if (
+        operation.operation == "drop"
+        and operation.target_type == "table"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return (
+            "drop-table",
+            TableResource(database=operation.database, table=operation.table),
+        )
+    if (
+        operation.operation in {"create", "drop"}
+        and operation.target_type == "index"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return (
+            "alter-table",
+            TableResource(database=operation.database, table=operation.table),
+        )
+    return None
+
+
+def operation_is_write(operation: Operation) -> bool:
+    return operation.operation in {
+        "insert",
+        "update",
+        "delete",
+        "create",
+        "alter",
+        "drop",
+        "begin",
+        "commit",
+        "rollback",
+        "attach",
+        "detach",
+        "pragma",
+        "analyze",
+        "reindex",
+    }
+
+
+async def ensure_query_write_permissions(
+    datasette: Datasette,
+    database: str,
+    sql: str,
+    *,
+    actor: dict[str, object] | None = None,
+    params: dict[str, object] | None = None,
+    analysis: SQLAnalysis | None = None,
+) -> SQLAnalysis:
     db = datasette.get_database(database)
     if analysis is None:
         if params is None:
@@ -606,18 +685,38 @@ async def ensure_query_write_permissions(
         except sqlite3.DatabaseError as ex:
             raise Forbidden(f"Could not analyze query: {ex}") from ex
 
-    for access in analysis.table_accesses:
-        action = write_actions.get(access.operation)
-        if action is None:
+    has_semantic_schema_operation = any(
+        operation.operation in {"create", "alter", "drop"}
+        and operation.target_type in {"table", "index", "view", "trigger"}
+        for operation in analysis.operations
+    )
+    for operation in analysis.operations:
+        if operation.internal and has_semantic_schema_operation:
             continue
-        if access.database != database:
+        if has_semantic_schema_operation and operation.operation in {
+            "read",
+            "insert",
+            "update",
+            "delete",
+            "reindex",
+        }:
+            continue
+        permission = permission_for_operation(operation)
+        if permission is None:
+            if operation_is_write(operation):
+                raise Forbidden(
+                    "Unsupported SQL operation: {} {}".format(
+                        operation.operation, operation.target_type
+                    )
+                )
+            continue
+        action, resource = permission
+        if operation.database != database:
             raise Forbidden("Writable queries may not write to attached databases")
         if not await datasette.allowed(
             action=action,
-            resource=TableResource(database=access.database, table=access.table),
+            resource=resource,
             actor=actor,
         ):
-            raise Forbidden(
-                f"Permission denied: need {action} on {access.database}/{access.table}"
-            )
+            raise Forbidden(f"Permission denied: need {action} on {resource}")
     return analysis
diff --git a/datasette/utils/sql_analysis.py b/datasette/utils/sql_analysis.py
index b5317b62..54f310fe 100644
--- a/datasette/utils/sql_analysis.py
+++ b/datasette/utils/sql_analysis.py
@@ -3,22 +3,66 @@ from typing import Literal
 
 from datasette.utils.sqlite import sqlite3
 
+SQLOperation = Literal[
+    "read",
+    "insert",
+    "update",
+    "delete",
+    "create",
+    "alter",
+    "drop",
+    "begin",
+    "commit",
+    "rollback",
+    "attach",
+    "detach",
+    "pragma",
+    "analyze",
+    "reindex",
+]
+SQLTargetType = Literal[
+    "table",
+    "index",
+    "view",
+    "trigger",
+    "schema",
+    "transaction",
+    "database",
+    "pragma",
+    "unknown",
+]
 SQLTableOperation = Literal["read", "insert", "update", "delete"]
 
 
 @dataclass(frozen=True)
-class SQLTableAccess:
-    operation: SQLTableOperation
+class Operation:
+    operation: SQLOperation
+    target_type: SQLTargetType
     database: str | None
-    table: str
+    table: str | None
     sqlite_schema: str | None
+    target: str | None = None
     columns: tuple[str, ...] = ()
     source: str | None = None
+    internal: bool = False
 
 
 @dataclass(frozen=True)
 class SQLAnalysis:
-    table_accesses: tuple[SQLTableAccess, ...]
+    operations: tuple[Operation, ...]
+
+
+# Hashable dict key for grouping repeated authorizer callbacks while collecting columns.
+@dataclass(frozen=True)
+class OperationKey:
+    operation: SQLOperation
+    target_type: SQLTargetType
+    database: str | None
+    table: str | None
+    sqlite_schema: str | None
+    target: str | None
+    source: str | None
+    internal: bool
 
 
 _ACTION_TO_OPERATION: dict[int, SQLTableOperation] = {
@@ -28,6 +72,36 @@ _ACTION_TO_OPERATION: dict[int, SQLTableOperation] = {
     sqlite3.SQLITE_DELETE: "delete",
 }
 
+# Values are (operation, target_type) pairs used to construct Operation objects.
+_CREATE_ACTIONS = {
+    sqlite3.SQLITE_CREATE_INDEX: ("create", "index"),
+    sqlite3.SQLITE_CREATE_TABLE: ("create", "table"),
+    sqlite3.SQLITE_CREATE_TRIGGER: ("create", "trigger"),
+    sqlite3.SQLITE_CREATE_VIEW: ("create", "view"),
+}
+_DROP_ACTIONS = {
+    sqlite3.SQLITE_DROP_INDEX: ("drop", "index"),
+    sqlite3.SQLITE_DROP_TABLE: ("drop", "table"),
+    sqlite3.SQLITE_DROP_TRIGGER: ("drop", "trigger"),
+    sqlite3.SQLITE_DROP_VIEW: ("drop", "view"),
+}
+for action_name, operation, target_type in (
+    ("SQLITE_CREATE_TEMP_INDEX", "create", "index"),
+    ("SQLITE_CREATE_TEMP_TABLE", "create", "table"),
+    ("SQLITE_CREATE_TEMP_TRIGGER", "create", "trigger"),
+    ("SQLITE_CREATE_TEMP_VIEW", "create", "view"),
+    ("SQLITE_DROP_TEMP_INDEX", "drop", "index"),
+    ("SQLITE_DROP_TEMP_TABLE", "drop", "table"),
+    ("SQLITE_DROP_TEMP_TRIGGER", "drop", "trigger"),
+    ("SQLITE_DROP_TEMP_VIEW", "drop", "view"),
+):
+    action_value = getattr(sqlite3, action_name, None)
+    if action_value is not None:
+        actions = _CREATE_ACTIONS if operation == "create" else _DROP_ACTIONS
+        actions[action_value] = (operation, target_type)
+
+_SQLITE_SCHEMA_TABLES = {"sqlite_master", "sqlite_schema"}
+
 
 def analyze_sql_tables(
     conn,
@@ -38,15 +112,13 @@ def analyze_sql_tables(
     schema_to_database: dict[str, str] | None = None,
 ) -> SQLAnalysis:
     """
-    Return tables accessed by a SQL statement according to SQLite's authorizer.
+    Return operations performed by a SQL statement according to SQLite's authorizer.
 
     This function is synchronous and connection-based. It temporarily installs a
-    SQLite authorizer, prepares ``EXPLAIN <sql>``, and returns the table access
+    SQLite authorizer, prepares ``EXPLAIN <sql>``, and returns the operation
     callbacks observed while SQLite compiles the statement.
     """
-    accesses: dict[
-        tuple[SQLTableOperation, str | None, str, str | None, str | None], set[str]
-    ] = {}
+    operations: dict[OperationKey, set[str]] = {}
 
     def database_for_schema(sqlite_schema):
         if schema_to_database and sqlite_schema in schema_to_database:
@@ -55,21 +127,166 @@ def analyze_sql_tables(
             return database_name
         return sqlite_schema
 
+    def record(
+        operation: SQLOperation,
+        target_type: SQLTargetType,
+        *,
+        database: str | None,
+        table: str | None,
+        sqlite_schema: str | None,
+        target: str | None,
+        source: str | None,
+        column: str | None = None,
+        internal: bool = False,
+    ):
+        key = OperationKey(
+            operation=operation,
+            target_type=target_type,
+            database=database,
+            table=table,
+            sqlite_schema=sqlite_schema,
+            target=target,
+            source=source,
+            internal=internal,
+        )
+        columns = operations.setdefault(key, set())
+        if column is not None:
+            columns.add(column)
+
     def authorizer(action, arg1, arg2, sqlite_schema, source):
         operation = _ACTION_TO_OPERATION.get(action)
-        if operation is None or arg1 is None:
+        if operation is not None and arg1 is not None:
+            target_type = "schema" if arg1 in _SQLITE_SCHEMA_TABLES else "table"
+            column = (
+                arg2 if operation in ("read", "update") and arg2 is not None else None
+            )
+            record(
+                operation,
+                target_type,
+                database=database_for_schema(sqlite_schema),
+                table=arg1 if target_type == "table" else None,
+                sqlite_schema=sqlite_schema,
+                target=arg1,
+                source=source,
+                column=column,
+                internal=target_type == "schema",
+            )
+            return sqlite3.SQLITE_OK
+
+        create_operation = _CREATE_ACTIONS.get(action)
+        if create_operation is not None and arg1 is not None:
+            operation, target_type = create_operation
+            related_table = arg2 if target_type in {"index", "trigger"} else arg1
+            record(
+                operation,
+                target_type,
+                database=database_for_schema(sqlite_schema),
+                table=related_table,
+                sqlite_schema=sqlite_schema,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        drop_operation = _DROP_ACTIONS.get(action)
+        if drop_operation is not None and arg1 is not None:
+            operation, target_type = drop_operation
+            related_table = arg2 if target_type in {"index", "trigger"} else arg1
+            record(
+                operation,
+                target_type,
+                database=database_for_schema(sqlite_schema),
+                table=related_table,
+                sqlite_schema=sqlite_schema,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_ALTER_TABLE and arg2 is not None:
+            record(
+                "alter",
+                "table",
+                database=database_for_schema(arg1),
+                table=arg2,
+                sqlite_schema=arg1,
+                target=arg2,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_TRANSACTION and arg1 is not None:
+            record(
+                arg1.lower(),
+                "transaction",
+                database=None,
+                table=None,
+                sqlite_schema=None,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_ATTACH and arg1 is not None:
+            record(
+                "attach",
+                "database",
+                database=None,
+                table=None,
+                sqlite_schema=None,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_DETACH and arg1 is not None:
+            record(
+                "detach",
+                "database",
+                database=None,
+                table=None,
+                sqlite_schema=None,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_PRAGMA and arg1 is not None:
+            record(
+                "pragma",
+                "pragma",
+                database=None,
+                table=None,
+                sqlite_schema=sqlite_schema,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_ANALYZE:
+            record(
+                "analyze",
+                "database" if arg1 is None else "table",
+                database=database_for_schema(sqlite_schema),
+                table=arg1,
+                sqlite_schema=sqlite_schema,
+                target=arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_REINDEX and arg1 is not None:
+            record(
+                "reindex",
+                "index",
+                database=database_for_schema(sqlite_schema),
+                table=None,
+                sqlite_schema=sqlite_schema,
+                target=arg1,
+                source=source,
+            )
             return sqlite3.SQLITE_OK
 
-        key = (
-            operation,
-            database_for_schema(sqlite_schema),
-            arg1,
-            sqlite_schema,
-            source,
-        )
-        columns = accesses.setdefault(key, set())
-        if operation in ("read", "update") and arg2 is not None:
-            columns.add(arg2)
         return sqlite3.SQLITE_OK
 
     conn.set_authorizer(authorizer)
@@ -78,22 +295,26 @@ def analyze_sql_tables(
     finally:
         conn.set_authorizer(None)
 
+    has_schema_operation = any(
+        key.target_type in {"table", "index", "view", "trigger"}
+        and key.operation in {"create", "alter", "drop"}
+        for key in operations
+    )
+
     return SQLAnalysis(
-        table_accesses=tuple(
-            SQLTableAccess(
-                operation=operation,
-                database=database,
-                table=table,
-                sqlite_schema=sqlite_schema,
+        operations=tuple(
+            Operation(
+                operation=key.operation,
+                target_type=key.target_type,
+                database=key.database,
+                table=key.table,
+                sqlite_schema=key.sqlite_schema,
+                target=key.target,
                 columns=tuple(sorted(columns)),
-                source=source,
+                source=key.source,
+                internal=key.internal
+                or (has_schema_operation and key.target_type == "schema"),
             )
-            for (
-                operation,
-                database,
-                table,
-                sqlite_schema,
-                source,
-            ), columns in accesses.items()
+            for key, columns in operations.items()
         )
     )
diff --git a/datasette/views/execute_write.py b/datasette/views/execute_write.py
index 0054300c..cead8926 100644
--- a/datasette/views/execute_write.py
+++ b/datasette/views/execute_write.py
@@ -193,9 +193,12 @@ class ExecuteWriteView(BaseView):
                 status=400,
             )
 
-        message = "Query executed, {} row{} affected".format(
-            cursor.rowcount, "" if cursor.rowcount == 1 else "s"
-        )
+        if cursor.rowcount == -1:
+            message = "Query executed"
+        else:
+            message = "Query executed, {} row{} affected".format(
+                cursor.rowcount, "" if cursor.rowcount == 1 else "s"
+            )
         if _wants_json(request, is_json, data):
             return _block_framing(
                 Response.json(
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index 46d71b8e..922f4e52 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -1,8 +1,12 @@
 import json
 import re
 
-from datasette.resources import DatabaseResource, TableResource
-from datasette.stored_queries import StoredQuery
+from datasette.resources import DatabaseResource
+from datasette.stored_queries import (
+    StoredQuery,
+    operation_is_write,
+    permission_for_operation,
+)
 from datasette.utils import (
     named_parameters as derive_named_parameters,
     escape_sqlite,
@@ -12,6 +16,7 @@ from datasette.utils import (
     InvalidSql,
 )
 from datasette.utils.asgi import Forbidden
+from datasette.utils.sql_analysis import Operation, SQLAnalysis
 
 _query_name_re = re.compile(r"^[^/\.\n]+$")
 
@@ -123,11 +128,8 @@ def _coerce_query_parameters(value, derived):
     return parameters
 
 
-def _analysis_is_write(analysis):
-    return any(
-        access.operation in {"insert", "update", "delete"}
-        for access in analysis.table_accesses
-    )
+def _analysis_is_write(analysis: SQLAnalysis) -> bool:
+    return any(operation_is_write(operation) for operation in analysis.operations)
 
 
 def _block_framing(response):
@@ -201,34 +203,66 @@ async def _analyze_user_query(datasette, db, sql, *, actor):
     return is_write, derived, analysis
 
 
-def _analysis_rows(analysis):
-    write_actions = {
-        "insert": "insert-row",
-        "update": "update-row",
-        "delete": "delete-row",
-    }
-    return [
-        {
-            "operation": access.operation,
-            "database": access.database,
-            "table": access.table,
-            "required_permission": write_actions.get(access.operation, ""),
-            "source": access.source,
-        }
-        for access in analysis.table_accesses
-    ]
+def _semantic_schema_operation_is_present(operations: tuple[Operation, ...]) -> bool:
+    return any(
+        operation.operation in {"create", "alter", "drop"}
+        and operation.target_type in {"table", "index", "view", "trigger"}
+        for operation in operations
+    )
 
 
-async def _analysis_rows_with_permissions(datasette, analysis, actor):
+def _display_operations(analysis: SQLAnalysis) -> list[Operation]:
+    has_semantic_schema_operation = _semantic_schema_operation_is_present(
+        analysis.operations
+    )
+    operations = []
+    for operation in analysis.operations:
+        if operation.internal and has_semantic_schema_operation:
+            continue
+        if has_semantic_schema_operation and operation.operation in {
+            "read",
+            "insert",
+            "update",
+            "delete",
+            "reindex",
+        }:
+            continue
+        operations.append(operation)
+    return operations
+
+
+def _analysis_rows(analysis: SQLAnalysis) -> list[dict[str, object]]:
+    rows = []
+    for operation in _display_operations(analysis):
+        permission = permission_for_operation(operation)
+        required_permission = permission[0] if permission else ""
+        rows.append(
+            {
+                "operation": operation.operation,
+                "database": operation.database,
+                "table": operation.table or operation.target,
+                "required_permission": required_permission,
+                "source": operation.source,
+            }
+        )
+    return rows
+
+
+async def _analysis_rows_with_permissions(
+    datasette, analysis: SQLAnalysis, actor
+) -> list[dict[str, object]]:
     rows = _analysis_rows(analysis)
-    for row in rows:
-        permission = row["required_permission"]
+    for row, operation in zip(rows, _display_operations(analysis)):
+        permission = permission_for_operation(operation)
         if permission:
+            action, resource = permission
             row["allowed"] = await datasette.allowed(
-                action=permission,
-                resource=TableResource(row["database"], row["table"]),
+                action=action,
+                resource=resource,
                 actor=actor,
             )
+        elif operation_is_write(operation):
+            row["allowed"] = False
         else:
             row["allowed"] = None
     return rows
@@ -398,15 +432,19 @@ async def _inserted_row_url(datasette, db, analysis, cursor):
     if lastrowid is None:
         return None
     direct_inserts = [
-        access
-        for access in analysis.table_accesses
-        if access.operation == "insert"
-        and access.source is None
-        and access.database == db.name
+        operation
+        for operation in analysis.operations
+        if operation.operation == "insert"
+        and operation.target_type == "table"
+        and not operation.internal
+        and operation.source is None
+        and operation.database == db.name
     ]
     if len(direct_inserts) != 1:
         return None
     table = direct_inserts[0].table
+    if table is None:
+        return None
     pks = await db.primary_keys(table)
     use_rowid = not pks
     select = (
diff --git a/tests/test_actions_sql.py b/tests/test_actions_sql.py
index 863d2529..a1fca971 100644
--- a/tests/test_actions_sql.py
+++ b/tests/test_actions_sql.py
@@ -12,10 +12,22 @@ import pytest
 import pytest_asyncio
 from datasette.app import Datasette
 from datasette.permissions import PermissionSQL
-from datasette.resources import TableResource
+from datasette.resources import DatabaseResource, QueryResource, TableResource
 from datasette import hookimpl
 
 
+def test_resource_string_representations():
+    assert str(DatabaseResource("content")) == "content"
+    assert repr(DatabaseResource("content")) == (
+        "DatabaseResource(parent='content', child=None)"
+    )
+    assert str(TableResource("content", "dogs")) == "content/dogs"
+    assert repr(TableResource("content", "dogs")) == (
+        "TableResource(parent='content', child='dogs')"
+    )
+    assert str(QueryResource("content", "insert-a-dog")) == "content/insert-a-dog"
+
+
 # Test plugin that provides permission rules
 class PermissionRulesPlugin:
     def __init__(self, rules_callback):
diff --git a/tests/test_internals_database.py b/tests/test_internals_database.py
index 5481a398..d6e130b4 100644
--- a/tests/test_internals_database.py
+++ b/tests/test_internals_database.py
@@ -698,14 +698,17 @@ async def test_analyze_sql():
 
     assert [
         (
-            access.operation,
-            access.database,
-            access.sqlite_schema,
-            access.table,
-            access.columns,
-            access.source,
+            operation.operation,
+            operation.database,
+            operation.sqlite_schema,
+            operation.table,
+            operation.columns,
+            operation.source,
         )
-        for access in analysis.table_accesses
+        for operation in analysis.operations
+        if operation.target_type == "table"
+        and operation.operation in {"read", "insert", "update", "delete"}
+        and not operation.internal
     ] == [
         ("read", "data", "main", "dogs", ("id", "name"), None),
     ]
@@ -722,14 +725,17 @@ async def test_analyze_sql_insert_select():
 
     assert {
         (
-            access.operation,
-            access.database,
-            access.sqlite_schema,
-            access.table,
-            access.columns,
-            access.source,
+            operation.operation,
+            operation.database,
+            operation.sqlite_schema,
+            operation.table,
+            operation.columns,
+            operation.source,
         )
-        for access in analysis.table_accesses
+        for operation in analysis.operations
+        if operation.target_type == "table"
+        and operation.operation in {"read", "insert", "update", "delete"}
+        and not operation.internal
     } == {
         ("insert", "data", "main", "dogs", (), None),
         ("read", "data", "main", "cats", ("name",), None),
diff --git a/tests/test_queries.py b/tests/test_queries.py
index 59fab8c0..4b8a6486 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -1643,6 +1643,172 @@ async def test_execute_write_post_requires_database_and_table_permissions():
     assert (await db.execute("select name from dogs")).first()[0] == "Cleo"
 
 
+@pytest.mark.asyncio
+async def test_execute_write_create_table_uses_create_table_permission():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "permissions": {
+                "insert-row": {"id": "row-writer"},
+                "update-row": {"id": "row-writer"},
+            },
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": ["creator", "row-writer"]},
+                        "execute-write-sql": {"id": ["creator", "row-writer"]},
+                        "create-table": {"id": "creator"},
+                    }
+                }
+            },
+        },
+    )
+    db = ds.add_memory_database("execute_write_create_table", name="data")
+    await ds.invoke_startup()
+
+    analysis_response = await ds.client.get(
+        "/data/-/execute-write/analyze",
+        actor={"id": "creator"},
+        params={"sql": "create table foobar (id integer primary key, name text)"},
+    )
+    allowed_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "creator"},
+        json={"sql": "create table foobar (id integer primary key, name text)"},
+    )
+    row_permission_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "row-writer"},
+        json={"sql": "create table should_not_exist (id integer primary key)"},
+    )
+
+    assert analysis_response.status_code == 200
+    analysis_data = analysis_response.json()
+    assert analysis_data["ok"] is True
+    assert analysis_data["execute_disabled"] is False
+    assert analysis_data["analysis_rows"] == [
+        {
+            "operation": "create",
+            "database": "data",
+            "table": "foobar",
+            "required_permission": "create-table",
+            "source": None,
+            "allowed": True,
+        }
+    ]
+
+    assert allowed_response.status_code == 200
+    assert allowed_response.json()["ok"] is True
+    assert allowed_response.json()["message"] == "Query executed"
+    assert await db.table_exists("foobar")
+
+    assert row_permission_response.status_code == 403
+    assert row_permission_response.json()["errors"] == [
+        "Permission denied: need create-table on data"
+    ]
+    assert not await db.table_exists("should_not_exist")
+
+
+@pytest.mark.asyncio
+async def test_execute_write_alter_and_drop_table_use_schema_permissions():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "permissions": {
+                "delete-row": {"id": "row-writer"},
+                "update-row": {"id": "row-writer"},
+            },
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": ["alterer", "dropper", "row-writer"]},
+                        "execute-write-sql": {
+                            "id": ["alterer", "dropper", "row-writer"]
+                        },
+                    },
+                    "tables": {
+                        "dogs": {
+                            "permissions": {
+                                "alter-table": {"id": "alterer"},
+                                "drop-table": {"id": "dropper"},
+                            }
+                        }
+                    },
+                }
+            },
+        },
+    )
+    db = ds.add_memory_database("execute_write_alter_drop_table", name="data")
+    await db.execute_write("create table dogs (id integer primary key, name text)")
+    await db.execute_write("create table cats (id integer primary key, name text)")
+    await ds.invoke_startup()
+
+    alter_allowed_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "alterer"},
+        json={"sql": "alter table dogs add column age integer"},
+    )
+    alter_row_permission_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "row-writer"},
+        json={"sql": "alter table cats add column age integer"},
+    )
+
+    assert alter_allowed_response.status_code == 200
+    assert "age" in [column.name for column in await db.table_column_details("dogs")]
+    assert alter_row_permission_response.status_code == 403
+    assert alter_row_permission_response.json()["errors"] == [
+        "Permission denied: need alter-table on data/cats"
+    ]
+    assert "age" not in [
+        column.name for column in await db.table_column_details("cats")
+    ]
+
+    create_index_allowed_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "alterer"},
+        json={"sql": "create index idx_dogs_name on dogs(name)"},
+    )
+    create_index_row_permission_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "row-writer"},
+        json={"sql": "create index idx_cats_name on cats(name)"},
+    )
+    drop_index_allowed_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "alterer"},
+        json={"sql": "drop index idx_dogs_name"},
+    )
+
+    assert create_index_allowed_response.status_code == 200
+    assert create_index_row_permission_response.status_code == 403
+    assert create_index_row_permission_response.json()["errors"] == [
+        "Permission denied: need alter-table on data/cats"
+    ]
+    assert drop_index_allowed_response.status_code == 200
+
+    drop_allowed_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "dropper"},
+        json={"sql": "drop table dogs"},
+    )
+    drop_row_permission_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "row-writer"},
+        json={"sql": "drop table cats"},
+    )
+
+    assert drop_allowed_response.status_code == 200
+    assert not await db.table_exists("dogs")
+    assert drop_row_permission_response.status_code == 403
+    assert drop_row_permission_response.json()["errors"] == [
+        "Permission denied: need drop-table on data/cats"
+    ]
+    assert await db.table_exists("cats")
+
+
 @pytest.mark.asyncio
 async def test_execute_write_insert_links_to_inserted_row():
     ds = Datasette(memory=True, default_deny=True)
diff --git a/tests/test_utils_sql_analysis.py b/tests/test_utils_sql_analysis.py
index 5730cd0d..5306a515 100644
--- a/tests/test_utils_sql_analysis.py
+++ b/tests/test_utils_sql_analysis.py
@@ -26,17 +26,20 @@ def conn():
         conn.close()
 
 
-def as_tuples(analysis):
+def table_operation_tuples(analysis):
     return [
         (
-            access.operation,
-            access.database,
-            access.sqlite_schema,
-            access.table,
-            access.columns,
-            access.source,
+            operation.operation,
+            operation.database,
+            operation.sqlite_schema,
+            operation.table,
+            operation.columns,
+            operation.source,
         )
-        for access in analysis.table_accesses
+        for operation in analysis.operations
+        if operation.target_type == "table"
+        and operation.operation in {"read", "insert", "update", "delete"}
+        and not operation.internal
     ]
 
 
@@ -48,7 +51,7 @@ def test_analyze_select_tables(conn):
         database_name="data",
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("read", "data", "main", "cats", ("id", "name"), None),
         ("read", "data", "main", "dogs", ("age", "id", "name"), None),
     }
@@ -57,11 +60,73 @@ def test_analyze_select_tables(conn):
 def test_analyze_uses_sqlite_schema_as_default_database(conn):
     analysis = analyze_sql_tables(conn, "select name from dogs")
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("read", "main", "main", "dogs", ("name",), None),
     }
 
 
+def operation_dict(operation):
+    return {
+        "operation": operation.operation,
+        "target_type": operation.target_type,
+        "database": operation.database,
+        "sqlite_schema": operation.sqlite_schema,
+        "table": operation.table,
+        "target": operation.target,
+        "columns": operation.columns,
+        "source": operation.source,
+        "internal": operation.internal,
+    }
+
+
+def test_analyze_create_table_operation():
+    conn = sqlite3.connect(":memory:")
+    try:
+        analysis = analyze_sql_tables(
+            conn,
+            "create table foobar (id integer primary key, name text)",
+            database_name="data",
+        )
+    finally:
+        conn.close()
+
+    assert {
+        "operation": "create",
+        "target_type": "table",
+        "database": "data",
+        "sqlite_schema": "main",
+        "table": "foobar",
+        "target": "foobar",
+        "columns": (),
+        "source": None,
+        "internal": False,
+    } in [operation_dict(operation) for operation in analysis.operations]
+    assert not [
+        operation
+        for operation in analysis.operations
+        if operation.table in {"sqlite_master", "sqlite_schema"}
+        and not operation.internal
+    ]
+
+
+def test_analyze_transaction_operation(conn):
+    analysis = analyze_sql_tables(conn, "commit", database_name="data")
+
+    assert [operation_dict(operation) for operation in analysis.operations] == [
+        {
+            "operation": "commit",
+            "target_type": "transaction",
+            "database": None,
+            "sqlite_schema": None,
+            "table": None,
+            "target": "COMMIT",
+            "columns": (),
+            "source": None,
+            "internal": False,
+        }
+    ]
+
+
 def test_analyze_insert_tables(conn):
     analysis = analyze_sql_tables(
         conn,
@@ -70,7 +135,7 @@ def test_analyze_insert_tables(conn):
         database_name="data",
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("insert", "data", "main", "dogs", (), None),
         ("read", "data", "main", "dogs", ("id", "name"), "dogs_after_insert"),
         ("update", "data", "main", "cats", ("name",), "dogs_after_insert"),
@@ -87,7 +152,7 @@ def test_analyze_update_tables(conn):
         database_name="data",
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("update", "data", "main", "dogs", ("age",), None),
         ("read", "data", "main", "dogs", ("age", "name"), None),
     }
@@ -101,7 +166,7 @@ def test_analyze_delete_tables(conn):
         database_name="data",
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("delete", "data", "main", "dogs", (), None),
         ("read", "data", "main", "dogs", ("name",), None),
     }
@@ -121,7 +186,7 @@ def test_analyze_insert_select_with_cte(conn):
         database_name="data",
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("insert", "data", "main", "cats", (), None),
         ("read", "data", "main", "dogs", ("age", "name"), "old_dogs"),
     }
@@ -135,7 +200,7 @@ def test_analyze_view_with_instead_of_trigger(conn):
         database_name="data",
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("update", "data", "main", "dog_names", ("name",), None),
         ("read", "data", "main", "dogs", ("id", "name"), "dog_names"),
         ("read", "data", "main", "dog_names", ("id", "name"), "dog_names"),
@@ -163,7 +228,7 @@ def test_analyze_attached_database_tables(conn):
         schema_to_database={"extra": "extra_db"},
     )
 
-    assert set(as_tuples(analysis)) == {
+    assert set(table_operation_tuples(analysis)) == {
         ("insert", "extra_db", "extra", "people", (), None),
         ("read", "data", "main", "dogs", ("name",), None),
     }

From 86d0e7335f98a88874df31ec0adb64967446dfac Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Wed, 27 May 2026 14:52:52 -0700
Subject: [PATCH 3/8] Deny unsupported write SQL operations by default

Require view-table permission for reads discovered inside write SQL analysis, including INSERT ... SELECT and CREATE TABLE ... AS SELECT.

Record additional SQLite authorizer callbacks as Operation values so unsupported functions, savepoints, virtual table DDL, and unknown callbacks are denied unless explicitly handled.
---
 datasette/stored_queries.py      |  43 +++----
 datasette/utils/sql_analysis.py  | 192 +++++++++++++++++++++++++++++--
 datasette/views/execute_write.py |   4 +-
 datasette/views/query_helpers.py |  32 ++----
 tests/test_queries.py            | 136 ++++++++++++++++++++--
 tests/test_utils_sql_analysis.py |  94 +++++++++++++++
 6 files changed, 433 insertions(+), 68 deletions(-)

diff --git a/datasette/stored_queries.py b/datasette/stored_queries.py
index c4b083e5..4b0fe6a6 100644
--- a/datasette/stored_queries.py
+++ b/datasette/stored_queries.py
@@ -592,6 +592,16 @@ PermissionRequirement = tuple[str, Resource]
 
 
 def permission_for_operation(operation: Operation) -> PermissionRequirement | None:
+    if (
+        operation.operation == "read"
+        and operation.target_type == "table"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return (
+            "view-table",
+            TableResource(database=operation.database, table=operation.table),
+        )
     write_actions = {
         "insert": "insert-row",
         "update": "update-row",
@@ -648,6 +658,10 @@ def permission_for_operation(operation: Operation) -> PermissionRequirement | No
     return None
 
 
+def operation_should_be_ignored(operation: Operation) -> bool:
+    return operation.internal or operation.operation == "select"
+
+
 def operation_is_write(operation: Operation) -> bool:
     return operation.operation in {
         "insert",
@@ -659,11 +673,13 @@ def operation_is_write(operation: Operation) -> bool:
         "begin",
         "commit",
         "rollback",
+        "savepoint",
         "attach",
         "detach",
         "pragma",
         "analyze",
         "reindex",
+        "unknown",
     }
 
 
@@ -685,34 +701,19 @@ async def ensure_query_write_permissions(
         except sqlite3.DatabaseError as ex:
             raise Forbidden(f"Could not analyze query: {ex}") from ex
 
-    has_semantic_schema_operation = any(
-        operation.operation in {"create", "alter", "drop"}
-        and operation.target_type in {"table", "index", "view", "trigger"}
-        for operation in analysis.operations
-    )
     for operation in analysis.operations:
-        if operation.internal and has_semantic_schema_operation:
-            continue
-        if has_semantic_schema_operation and operation.operation in {
-            "read",
-            "insert",
-            "update",
-            "delete",
-            "reindex",
-        }:
+        if operation_should_be_ignored(operation):
             continue
         permission = permission_for_operation(operation)
         if permission is None:
-            if operation_is_write(operation):
-                raise Forbidden(
-                    "Unsupported SQL operation: {} {}".format(
-                        operation.operation, operation.target_type
-                    )
+            raise Forbidden(
+                "Unsupported SQL operation: {} {}".format(
+                    operation.operation, operation.target_type
                 )
-            continue
+            )
         action, resource = permission
         if operation.database != database:
-            raise Forbidden("Writable queries may not write to attached databases")
+            raise Forbidden("Writable queries may not access attached databases")
         if not await datasette.allowed(
             action=action,
             resource=resource,
diff --git a/datasette/utils/sql_analysis.py b/datasette/utils/sql_analysis.py
index 54f310fe..8963da77 100644
--- a/datasette/utils/sql_analysis.py
+++ b/datasette/utils/sql_analysis.py
@@ -8,30 +8,39 @@ SQLOperation = Literal[
     "insert",
     "update",
     "delete",
+    "select",
+    "function",
     "create",
     "alter",
     "drop",
     "begin",
     "commit",
     "rollback",
+    "savepoint",
     "attach",
     "detach",
     "pragma",
     "analyze",
     "reindex",
+    "unknown",
 ]
 SQLTargetType = Literal[
     "table",
     "index",
     "view",
     "trigger",
+    "virtual-table",
     "schema",
+    "statement",
     "transaction",
     "database",
     "pragma",
+    "function",
     "unknown",
 ]
 SQLTableOperation = Literal["read", "insert", "update", "delete"]
+SQLSchemaOperation = Literal["create", "drop"]
+SQLSchemaTargetType = Literal["index", "table", "trigger", "view", "virtual-table"]
 
 
 @dataclass(frozen=True)
@@ -73,19 +82,34 @@ _ACTION_TO_OPERATION: dict[int, SQLTableOperation] = {
 }
 
 # Values are (operation, target_type) pairs used to construct Operation objects.
-_CREATE_ACTIONS = {
+_CREATE_ACTIONS: dict[int, tuple[SQLSchemaOperation, SQLSchemaTargetType]] = {
     sqlite3.SQLITE_CREATE_INDEX: ("create", "index"),
     sqlite3.SQLITE_CREATE_TABLE: ("create", "table"),
     sqlite3.SQLITE_CREATE_TRIGGER: ("create", "trigger"),
     sqlite3.SQLITE_CREATE_VIEW: ("create", "view"),
 }
-_DROP_ACTIONS = {
+_DROP_ACTIONS: dict[int, tuple[SQLSchemaOperation, SQLSchemaTargetType]] = {
     sqlite3.SQLITE_DROP_INDEX: ("drop", "index"),
     sqlite3.SQLITE_DROP_TABLE: ("drop", "table"),
     sqlite3.SQLITE_DROP_TRIGGER: ("drop", "trigger"),
     sqlite3.SQLITE_DROP_VIEW: ("drop", "view"),
 }
-for action_name, operation, target_type in (
+
+
+def _add_schema_action(
+    action_name: str,
+    operation: SQLSchemaOperation,
+    target_type: SQLSchemaTargetType,
+) -> None:
+    action_value = getattr(sqlite3, action_name, None)
+    if action_value is not None:
+        actions = _CREATE_ACTIONS if operation == "create" else _DROP_ACTIONS
+        actions[action_value] = (operation, target_type)
+
+
+_TEMP_SCHEMA_ACTIONS: tuple[
+    tuple[str, SQLSchemaOperation, SQLSchemaTargetType], ...
+] = (
     ("SQLITE_CREATE_TEMP_INDEX", "create", "index"),
     ("SQLITE_CREATE_TEMP_TABLE", "create", "table"),
     ("SQLITE_CREATE_TEMP_TRIGGER", "create", "trigger"),
@@ -94,13 +118,76 @@ for action_name, operation, target_type in (
     ("SQLITE_DROP_TEMP_TABLE", "drop", "table"),
     ("SQLITE_DROP_TEMP_TRIGGER", "drop", "trigger"),
     ("SQLITE_DROP_TEMP_VIEW", "drop", "view"),
-):
-    action_value = getattr(sqlite3, action_name, None)
-    if action_value is not None:
-        actions = _CREATE_ACTIONS if operation == "create" else _DROP_ACTIONS
-        actions[action_value] = (operation, target_type)
+)
+for schema_action in _TEMP_SCHEMA_ACTIONS:
+    _add_schema_action(*schema_action)
 
-_SQLITE_SCHEMA_TABLES = {"sqlite_master", "sqlite_schema"}
+_VTABLE_SCHEMA_ACTIONS: tuple[
+    tuple[str, SQLSchemaOperation, SQLSchemaTargetType], ...
+] = (
+    ("SQLITE_CREATE_VTABLE", "create", "virtual-table"),
+    ("SQLITE_DROP_VTABLE", "drop", "virtual-table"),
+)
+for schema_action in _VTABLE_SCHEMA_ACTIONS:
+    _add_schema_action(*schema_action)
+
+_SQLITE_SCHEMA_TABLES = {
+    "sqlite_master",
+    "sqlite_schema",
+    "sqlite_temp_master",
+    "sqlite_temp_schema",
+}
+_SQLITE_INTERNAL_SCHEMA_FUNCTIONS = {
+    "length",
+    "like",
+    "printf",
+    "sqlite_drop_column",
+    "sqlite_rename_column",
+    "sqlite_rename_quotefix",
+    "sqlite_rename_table",
+    "sqlite_rename_test",
+    "substr",
+}
+
+_AUTHORIZER_ACTION_NAMES = {
+    getattr(sqlite3, name): name
+    for name in (
+        "SQLITE_CREATE_INDEX",
+        "SQLITE_CREATE_TABLE",
+        "SQLITE_CREATE_TEMP_INDEX",
+        "SQLITE_CREATE_TEMP_TABLE",
+        "SQLITE_CREATE_TEMP_TRIGGER",
+        "SQLITE_CREATE_TEMP_VIEW",
+        "SQLITE_CREATE_TRIGGER",
+        "SQLITE_CREATE_VIEW",
+        "SQLITE_DELETE",
+        "SQLITE_DROP_INDEX",
+        "SQLITE_DROP_TABLE",
+        "SQLITE_DROP_TEMP_INDEX",
+        "SQLITE_DROP_TEMP_TABLE",
+        "SQLITE_DROP_TEMP_TRIGGER",
+        "SQLITE_DROP_TEMP_VIEW",
+        "SQLITE_DROP_TRIGGER",
+        "SQLITE_DROP_VIEW",
+        "SQLITE_INSERT",
+        "SQLITE_PRAGMA",
+        "SQLITE_READ",
+        "SQLITE_SELECT",
+        "SQLITE_TRANSACTION",
+        "SQLITE_UPDATE",
+        "SQLITE_ATTACH",
+        "SQLITE_DETACH",
+        "SQLITE_ALTER_TABLE",
+        "SQLITE_REINDEX",
+        "SQLITE_ANALYZE",
+        "SQLITE_CREATE_VTABLE",
+        "SQLITE_DROP_VTABLE",
+        "SQLITE_FUNCTION",
+        "SQLITE_SAVEPOINT",
+        "SQLITE_RECURSIVE",
+    )
+    if hasattr(sqlite3, name)
+}
 
 
 def analyze_sql_tables(
@@ -287,6 +374,52 @@ def analyze_sql_tables(
             )
             return sqlite3.SQLITE_OK
 
+        if action == sqlite3.SQLITE_SELECT:
+            record(
+                "select",
+                "statement",
+                database=None,
+                table=None,
+                sqlite_schema=sqlite_schema,
+                target=None,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_FUNCTION and arg2 is not None:
+            record(
+                "function",
+                "function",
+                database=None,
+                table=None,
+                sqlite_schema=sqlite_schema,
+                target=arg2,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        if action == sqlite3.SQLITE_SAVEPOINT and arg1 is not None:
+            record(
+                "savepoint",
+                "transaction",
+                database=None,
+                table=None,
+                sqlite_schema=sqlite_schema,
+                target="{} {}".format(arg1, arg2) if arg2 is not None else arg1,
+                source=source,
+            )
+            return sqlite3.SQLITE_OK
+
+        action_name = _AUTHORIZER_ACTION_NAMES.get(action, "SQLITE_{}".format(action))
+        record(
+            "unknown",
+            "unknown",
+            database=database_for_schema(sqlite_schema),
+            table=None,
+            sqlite_schema=sqlite_schema,
+            target=action_name,
+            source=source,
+        )
         return sqlite3.SQLITE_OK
 
     conn.set_authorizer(authorizer)
@@ -296,10 +429,46 @@ def analyze_sql_tables(
         conn.set_authorizer(None)
 
     has_schema_operation = any(
-        key.target_type in {"table", "index", "view", "trigger"}
+        key.target_type in {"table", "index", "view", "trigger", "virtual-table"}
         and key.operation in {"create", "alter", "drop"}
         for key in operations
     )
+    dropped_tables = {
+        (key.database, key.table)
+        for key in operations
+        if key.operation == "drop" and key.target_type == "table"
+    }
+
+    def key_is_drop_table_delete(key: OperationKey) -> bool:
+        return (
+            key.operation == "delete"
+            and key.target_type == "table"
+            and (key.database, key.table) in dropped_tables
+        )
+
+    has_user_table_access_in_schema_operation = any(
+        key.operation in {"read", "insert", "update", "delete"}
+        and key.target_type == "table"
+        and not key.internal
+        and not key_is_drop_table_delete(key)
+        for key in operations
+    )
+
+    def operation_is_internal(key: OperationKey) -> bool:
+        if key.internal or (has_schema_operation and key.target_type == "schema"):
+            return True
+        if has_schema_operation and key.operation == "reindex":
+            return True
+        if (
+            has_schema_operation
+            and not has_user_table_access_in_schema_operation
+            and key.operation == "function"
+            and key.target in _SQLITE_INTERNAL_SCHEMA_FUNCTIONS
+        ):
+            return True
+        if key_is_drop_table_delete(key):
+            return True
+        return False
 
     return SQLAnalysis(
         operations=tuple(
@@ -312,8 +481,7 @@ def analyze_sql_tables(
                 target=key.target,
                 columns=tuple(sorted(columns)),
                 source=key.source,
-                internal=key.internal
-                or (has_schema_operation and key.target_type == "schema"),
+                internal=operation_is_internal(key),
             )
             for key, columns in operations.items()
         )
diff --git a/datasette/views/execute_write.py b/datasette/views/execute_write.py
index cead8926..19006ac5 100644
--- a/datasette/views/execute_write.py
+++ b/datasette/views/execute_write.py
@@ -99,9 +99,7 @@ class ExecuteWriteView(BaseView):
                 "parameter_names": parameter_names,
                 "parameter_values": parameter_values,
                 "analysis_error": analysis_error,
-                "analysis_rows": [
-                    row for row in analysis_rows if row["operation"] != "read"
-                ],
+                "analysis_rows": analysis_rows,
                 "execution_message": execution_message,
                 "execution_links": execution_links,
                 "execution_ok": execution_ok,
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index 922f4e52..05a0d73e 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -5,6 +5,7 @@ from datasette.resources import DatabaseResource
 from datasette.stored_queries import (
     StoredQuery,
     operation_is_write,
+    operation_should_be_ignored,
     permission_for_operation,
 )
 from datasette.utils import (
@@ -203,29 +204,10 @@ async def _analyze_user_query(datasette, db, sql, *, actor):
     return is_write, derived, analysis
 
 
-def _semantic_schema_operation_is_present(operations: tuple[Operation, ...]) -> bool:
-    return any(
-        operation.operation in {"create", "alter", "drop"}
-        and operation.target_type in {"table", "index", "view", "trigger"}
-        for operation in operations
-    )
-
-
 def _display_operations(analysis: SQLAnalysis) -> list[Operation]:
-    has_semantic_schema_operation = _semantic_schema_operation_is_present(
-        analysis.operations
-    )
     operations = []
     for operation in analysis.operations:
-        if operation.internal and has_semantic_schema_operation:
-            continue
-        if has_semantic_schema_operation and operation.operation in {
-            "read",
-            "insert",
-            "update",
-            "delete",
-            "reindex",
-        }:
+        if operation_should_be_ignored(operation):
             continue
         operations.append(operation)
     return operations
@@ -252,6 +234,7 @@ async def _analysis_rows_with_permissions(
     datasette, analysis: SQLAnalysis, actor
 ) -> list[dict[str, object]]:
     rows = _analysis_rows(analysis)
+    is_write = _analysis_is_write(analysis)
     for row, operation in zip(rows, _display_operations(analysis)):
         permission = permission_for_operation(operation)
         if permission:
@@ -261,7 +244,7 @@ async def _analysis_rows_with_permissions(
                 resource=resource,
                 actor=actor,
             )
-        elif operation_is_write(operation):
+        elif is_write:
             row["allowed"] = False
         else:
             row["allowed"] = None
@@ -360,7 +343,7 @@ async def _execute_write_analysis_data(datasette, db, sql, actor):
         "ok": analysis_error is None,
         "parameters": parameter_names,
         "analysis_error": analysis_error,
-        "analysis_rows": [row for row in analysis_rows if row["operation"] != "read"],
+        "analysis_rows": analysis_rows,
         "execute_disabled": bool(
             (not sql)
             or analysis_error
@@ -374,6 +357,7 @@ async def _query_create_analysis_data(datasette, db, sql, actor):
     parameter_names = []
     analysis_rows = []
     analysis_error = None
+    analysis: SQLAnalysis | None = None
     if has_sql:
         try:
             parameter_names = _derived_query_parameters(sql)
@@ -390,9 +374,7 @@ async def _query_create_analysis_data(datasette, db, sql, actor):
         "analysis_error": analysis_error,
         "analysis_rows": analysis_rows,
         "has_sql": has_sql,
-        "analysis_is_write": bool(
-            analysis_rows and any(row["required_permission"] for row in analysis_rows)
-        ),
+        "analysis_is_write": _analysis_is_write(analysis) if analysis else False,
         "save_disabled": bool(
             (not has_sql)
             or analysis_error
diff --git a/tests/test_queries.py b/tests/test_queries.py
index 4b8a6486..97ec973f 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -1181,11 +1181,10 @@ async def test_create_query_ui_and_arbitrary_sql_save_link():
     assert '<th scope="col">Required permission</th>' in create_response.text
     assert '<th scope="col">Source</th>' not in create_response.text
     assert "<td><code>read</code></td>" in create_response.text
+    assert "<td><code>view-table</code></td>" in create_response.text
     assert (
-        create_response.text.count(
-            '<td><span class="execute-write-analysis-na">n/a</span></td>'
-        )
-        == 2
+        '<td><span class="execute-write-analysis-na">n/a</span></td>'
+        not in create_response.text
     )
     assert create_response.text.index(
         'value="Save query"'
@@ -1255,9 +1254,9 @@ async def test_create_query_analyze_endpoint_uses_sql_only():
             "operation": "read",
             "database": "data",
             "table": "dogs",
-            "required_permission": "",
+            "required_permission": "view-table",
             "source": None,
-            "allowed": None,
+            "allowed": True,
         }
     ]
 
@@ -1375,7 +1374,8 @@ async def test_execute_write_get_prepopulates_without_executing():
     assert '<th scope="col">Required permission</th>' in response.text
     assert "<td><code>insert</code></td>" in response.text
     assert "<td><code>update</code></td>" in response.text
-    assert "<td><code>read</code></td>" not in response.text
+    assert "<td><code>read</code></td>" in response.text
+    assert "<td><code>view-table</code></td>" in response.text
     assert 'action="/data/-/execute-write"' in response.text
     assert "insert into dogs (name) values (&#39;Cleo&#39;)" in response.text
     assert (await db.execute("select count(*) from dogs")).first()[0] == 0
@@ -1643,6 +1643,127 @@ async def test_execute_write_post_requires_database_and_table_permissions():
     assert (await db.execute("select name from dogs")).first()[0] == "Cleo"
 
 
+@pytest.mark.asyncio
+async def test_execute_write_insert_select_requires_view_table_on_source():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    },
+                    "tables": {
+                        "secret": {
+                            "permissions": {"view-table": {"id": "someone-else"}}
+                        },
+                        "public_log": {"permissions": {"insert-row": {"id": "writer"}}},
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_insert_select_source", name="data")
+    await db.execute_write("create table secret (value text)")
+    await db.execute_write("create table public_log (value text)")
+    await db.execute_write("insert into secret values ('sensitive')")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={"sql": "insert into public_log(value) select value from secret"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Permission denied: need view-table on data/secret"
+    ]
+    assert (await db.execute("select value from public_log")).dicts() == []
+
+
+@pytest.mark.asyncio
+async def test_execute_write_create_table_as_select_requires_view_table_on_source():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "creator"},
+                        "execute-write-sql": {"id": "creator"},
+                        "create-table": {"id": "creator"},
+                    },
+                    "tables": {
+                        "secret": {
+                            "permissions": {"view-table": {"id": "someone-else"}}
+                        }
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_create_as_select_source", name="data")
+    await db.execute_write("create table secret (value text)")
+    await db.execute_write("insert into secret values ('sensitive')")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "creator"},
+        json={"sql": "create table copied_secret as select value from secret"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Permission denied: need view-table on data/secret"
+    ]
+    assert not await db.table_exists("copied_secret")
+
+
+@pytest.mark.asyncio
+async def test_execute_write_rejects_function_operations():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    },
+                    "tables": {
+                        "dogs": {
+                            "permissions": {
+                                "insert-row": {"id": "writer"},
+                            }
+                        }
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_function_operation", name="data")
+    await db.execute_write("create table dogs (id integer primary key, name text)")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={"sql": "insert into dogs (name) values (upper('cleo'))"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Unsupported SQL operation: function function"
+    ]
+    assert (await db.execute("select name from dogs")).dicts() == []
+
+
 @pytest.mark.asyncio
 async def test_execute_write_create_table_uses_create_table_permission():
     ds = Datasette(
@@ -1733,6 +1854,7 @@ async def test_execute_write_alter_and_drop_table_use_schema_permissions():
                             "permissions": {
                                 "alter-table": {"id": "alterer"},
                                 "drop-table": {"id": "dropper"},
+                                "view-table": {"id": "alterer"},
                             }
                         }
                     },
diff --git a/tests/test_utils_sql_analysis.py b/tests/test_utils_sql_analysis.py
index 5306a515..2ae11502 100644
--- a/tests/test_utils_sql_analysis.py
+++ b/tests/test_utils_sql_analysis.py
@@ -127,6 +127,100 @@ def test_analyze_transaction_operation(conn):
     ]
 
 
+def test_analyze_savepoint_operation(conn):
+    analysis = analyze_sql_tables(conn, "savepoint s", database_name="data")
+
+    assert [operation_dict(operation) for operation in analysis.operations] == [
+        {
+            "operation": "savepoint",
+            "target_type": "transaction",
+            "database": None,
+            "sqlite_schema": None,
+            "table": None,
+            "target": "BEGIN s",
+            "columns": (),
+            "source": None,
+            "internal": False,
+        }
+    ]
+
+
+def test_analyze_function_operation(conn):
+    analysis = analyze_sql_tables(
+        conn,
+        "insert into dogs (name) values (upper(:name))",
+        {"name": "Cleo"},
+        database_name="data",
+    )
+
+    assert {
+        (
+            operation.operation,
+            operation.target_type,
+            operation.target,
+            operation.database,
+            operation.table,
+        )
+        for operation in analysis.operations
+    } == {
+        ("insert", "table", "dogs", "data", "dogs"),
+        ("function", "function", "upper", None, None),
+        ("read", "table", "dogs", "data", "dogs"),
+        ("update", "table", "cats", "data", "cats"),
+        ("read", "table", "cats", "data", "cats"),
+        ("insert", "table", "log", "data", "log"),
+    }
+
+
+def test_analyze_create_virtual_table_operation():
+    conn = sqlite3.connect(":memory:")
+    try:
+        analysis = analyze_sql_tables(
+            conn,
+            "create virtual table docs using fts5(body)",
+            database_name="data",
+        )
+    finally:
+        conn.close()
+
+    assert {
+        "operation": "create",
+        "target_type": "virtual-table",
+        "database": "data",
+        "sqlite_schema": "main",
+        "table": "docs",
+        "target": "docs",
+        "columns": (),
+        "source": None,
+        "internal": False,
+    } in [operation_dict(operation) for operation in analysis.operations]
+
+
+def test_analyze_create_table_as_select_function_is_not_internal():
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute("create table secret(value text)")
+        analysis = analyze_sql_tables(
+            conn,
+            "create table copied as select substr(value, 1, 1) from secret",
+            database_name="data",
+        )
+    finally:
+        conn.close()
+
+    assert {
+        "operation": "function",
+        "target_type": "function",
+        "database": None,
+        "sqlite_schema": None,
+        "table": None,
+        "target": "substr",
+        "columns": (),
+        "source": None,
+        "internal": False,
+    } in [operation_dict(operation) for operation in analysis.operations]
+
+
 def test_analyze_insert_tables(conn):
     analysis = analyze_sql_tables(
         conn,

From 03b2c66f6312b8317d87eb4c1326977f6f63b26d Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Wed, 27 May 2026 15:17:10 -0700
Subject: [PATCH 4/8] Require full row mutation permissions for raw SQL

Raw SQL insert and update statements can have broader effects than their SQLite authorizer callbacks reveal. INSERT OR REPLACE and UPDATE OR REPLACE can delete conflicting rows while only surfacing insert or update operations.

Expand table insert and update operations to require insert-row, update-row, and delete-row together. Keep delete operations mapped to delete-row, and update the analysis UI/API to report and evaluate multiple required permissions for a single operation.

Refs https://github.com/simonw/datasette/pull/2749#issuecomment-4559083539
---
 datasette/stored_queries.py      | 108 ++++++++++++-----
 datasette/views/query_helpers.py |  27 +++--
 tests/test_queries.py            | 200 ++++++++++++++++++++++++++++++-
 3 files changed, 290 insertions(+), 45 deletions(-)

diff --git a/datasette/stored_queries.py b/datasette/stored_queries.py
index 4b0fe6a6..cf44a9ff 100644
--- a/datasette/stored_queries.py
+++ b/datasette/stored_queries.py
@@ -588,10 +588,25 @@ async def list_queries(
     )
 
 
-PermissionRequirement = tuple[str, Resource]
+@dataclass(frozen=True)
+class PermissionRequirement:
+    action: str
+    resource: Resource
 
 
-def permission_for_operation(operation: Operation) -> PermissionRequirement | None:
+def row_mutation_requirements(
+    database: str, table: str
+) -> tuple[PermissionRequirement, ...]:
+    resource = TableResource(database=database, table=table)
+    return tuple(
+        PermissionRequirement(action=action, resource=resource)
+        for action in ("insert-row", "update-row", "delete-row")
+    )
+
+
+def permission_requirements_for_operation(
+    operation: Operation,
+) -> tuple[PermissionRequirement, ...]:
     if (
         operation.operation == "read"
         and operation.target_type == "table"
@@ -599,31 +614,45 @@ def permission_for_operation(operation: Operation) -> PermissionRequirement | No
         and operation.table is not None
     ):
         return (
-            "view-table",
-            TableResource(database=operation.database, table=operation.table),
+            PermissionRequirement(
+                action="view-table",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
         )
-    write_actions = {
-        "insert": "insert-row",
-        "update": "update-row",
-        "delete": "delete-row",
-    }
-    action = write_actions.get(operation.operation)
     if (
-        action
+        operation.operation in {"insert", "update"}
+        and operation.target_type == "table"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return row_mutation_requirements(
+            database=operation.database,
+            table=operation.table,
+        )
+    if (
+        operation.operation == "delete"
         and operation.target_type == "table"
         and operation.database is not None
         and operation.table is not None
     ):
         return (
-            action,
-            TableResource(database=operation.database, table=operation.table),
+            PermissionRequirement(
+                action="delete-row",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
         )
     if operation.operation == "create" and operation.target_type == "table":
         if operation.database is None:
-            return None
+            return ()
         return (
-            "create-table",
-            DatabaseResource(database=operation.database),
+            PermissionRequirement(
+                action="create-table",
+                resource=DatabaseResource(database=operation.database),
+            ),
         )
     if (
         operation.operation == "alter"
@@ -632,8 +661,12 @@ def permission_for_operation(operation: Operation) -> PermissionRequirement | No
         and operation.table is not None
     ):
         return (
-            "alter-table",
-            TableResource(database=operation.database, table=operation.table),
+            PermissionRequirement(
+                action="alter-table",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
         )
     if (
         operation.operation == "drop"
@@ -642,8 +675,12 @@ def permission_for_operation(operation: Operation) -> PermissionRequirement | No
         and operation.table is not None
     ):
         return (
-            "drop-table",
-            TableResource(database=operation.database, table=operation.table),
+            PermissionRequirement(
+                action="drop-table",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
         )
     if (
         operation.operation in {"create", "drop"}
@@ -652,10 +689,14 @@ def permission_for_operation(operation: Operation) -> PermissionRequirement | No
         and operation.table is not None
     ):
         return (
-            "alter-table",
-            TableResource(database=operation.database, table=operation.table),
+            PermissionRequirement(
+                action="alter-table",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
         )
-    return None
+    return ()
 
 
 def operation_should_be_ignored(operation: Operation) -> bool:
@@ -704,20 +745,23 @@ async def ensure_query_write_permissions(
     for operation in analysis.operations:
         if operation_should_be_ignored(operation):
             continue
-        permission = permission_for_operation(operation)
-        if permission is None:
+        permissions = permission_requirements_for_operation(operation)
+        if not permissions:
             raise Forbidden(
                 "Unsupported SQL operation: {} {}".format(
                     operation.operation, operation.target_type
                 )
             )
-        action, resource = permission
         if operation.database != database:
             raise Forbidden("Writable queries may not access attached databases")
-        if not await datasette.allowed(
-            action=action,
-            resource=resource,
-            actor=actor,
-        ):
-            raise Forbidden(f"Permission denied: need {action} on {resource}")
+        for permission in permissions:
+            if not await datasette.allowed(
+                action=permission.action,
+                resource=permission.resource,
+                actor=actor,
+            ):
+                raise Forbidden(
+                    f"Permission denied: need {permission.action} "
+                    f"on {permission.resource}"
+                )
     return analysis
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index 05a0d73e..7f3ef1bc 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -6,7 +6,7 @@ from datasette.stored_queries import (
     StoredQuery,
     operation_is_write,
     operation_should_be_ignored,
-    permission_for_operation,
+    permission_requirements_for_operation,
 )
 from datasette.utils import (
     named_parameters as derive_named_parameters,
@@ -216,8 +216,10 @@ def _display_operations(analysis: SQLAnalysis) -> list[Operation]:
 def _analysis_rows(analysis: SQLAnalysis) -> list[dict[str, object]]:
     rows = []
     for operation in _display_operations(analysis):
-        permission = permission_for_operation(operation)
-        required_permission = permission[0] if permission else ""
+        permissions = permission_requirements_for_operation(operation)
+        required_permission = ", ".join(
+            permission.action for permission in permissions
+        )
         rows.append(
             {
                 "operation": operation.operation,
@@ -236,14 +238,17 @@ async def _analysis_rows_with_permissions(
     rows = _analysis_rows(analysis)
     is_write = _analysis_is_write(analysis)
     for row, operation in zip(rows, _display_operations(analysis)):
-        permission = permission_for_operation(operation)
-        if permission:
-            action, resource = permission
-            row["allowed"] = await datasette.allowed(
-                action=action,
-                resource=resource,
-                actor=actor,
-            )
+        permissions = permission_requirements_for_operation(operation)
+        if permissions:
+            row["allowed"] = True
+            for permission in permissions:
+                if not await datasette.allowed(
+                    action=permission.action,
+                    resource=permission.resource,
+                    actor=actor,
+                ):
+                    row["allowed"] = False
+                    break
         elif is_write:
             row["allowed"] = False
         else:
diff --git a/tests/test_queries.py b/tests/test_queries.py
index 97ec973f..fcd19d1c 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -508,6 +508,8 @@ async def test_analyze_write_query_requires_table_permissions():
                     "dogs": {
                         "permissions": {
                             "insert-row": {"id": "writer"},
+                            "update-row": {"id": "writer"},
+                            "delete-row": {"id": "writer"},
                         }
                     }
                 }
@@ -1429,7 +1431,7 @@ async def test_execute_write_analyze_endpoint_uses_sql_only():
             "operation": "insert",
             "database": "data",
             "table": "dogs",
-            "required_permission": "insert-row",
+            "required_permission": "insert-row, update-row, delete-row",
             "source": None,
             "allowed": True,
         }
@@ -1627,6 +1629,40 @@ async def test_execute_write_post_requires_database_and_table_permissions():
             }
         }
     }
+    missing_update_permission = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={
+            "sql": "insert into dogs (name) values (:name)",
+            "params": {"name": "Cleo"},
+        },
+    )
+
+    assert missing_update_permission.status_code == 403
+    assert missing_update_permission.json()["errors"] == [
+        "Permission denied: need update-row on data/dogs"
+    ]
+
+    ds.config["databases"]["data"]["tables"]["dogs"]["permissions"][
+        "update-row"
+    ] = {"id": "writer"}
+    missing_delete_permission = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={
+            "sql": "insert into dogs (name) values (:name)",
+            "params": {"name": "Cleo"},
+        },
+    )
+
+    assert missing_delete_permission.status_code == 403
+    assert missing_delete_permission.json()["errors"] == [
+        "Permission denied: need delete-row on data/dogs"
+    ]
+
+    ds.config["databases"]["data"]["tables"]["dogs"]["permissions"][
+        "delete-row"
+    ] = {"id": "writer"}
     allowed = await ds.client.post(
         "/data/-/execute-write",
         actor={"id": "writer"},
@@ -1643,6 +1679,156 @@ async def test_execute_write_post_requires_database_and_table_permissions():
     assert (await db.execute("select name from dogs")).first()[0] == "Cleo"
 
 
+@pytest.mark.asyncio
+async def test_execute_write_insert_or_replace_requires_delete_row_permission():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    },
+                    "tables": {
+                        "users": {
+                            "permissions": {
+                                "insert-row": {"id": "writer"},
+                                "update-row": {"id": "writer"},
+                                "view-table": {"id": "writer"},
+                            }
+                        }
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_insert_or_replace", name="data")
+    await db.execute_write(
+        "create table users (id integer primary key, email text unique)"
+    )
+    await db.execute_write(
+        "insert into users (id, email) values "
+        "(1, 'a@example.com'), (2, 'b@example.com')"
+    )
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={
+            "sql": (
+                "insert or replace into users(id, email) "
+                "values (3, 'b@example.com')"
+            )
+        },
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Permission denied: need delete-row on data/users"
+    ]
+    assert (await db.execute("select id, email from users order by id")).dicts() == [
+        {"id": 1, "email": "a@example.com"},
+        {"id": 2, "email": "b@example.com"},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_execute_write_update_or_replace_requires_delete_row_permission():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    },
+                    "tables": {
+                        "users": {
+                            "permissions": {
+                                "insert-row": {"id": "writer"},
+                                "update-row": {"id": "writer"},
+                                "view-table": {"id": "writer"},
+                            }
+                        }
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_update_or_replace", name="data")
+    await db.execute_write(
+        "create table users (id integer primary key, email text unique)"
+    )
+    await db.execute_write(
+        "insert into users (id, email) values "
+        "(1, 'a@example.com'), (2, 'b@example.com')"
+    )
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={"sql": "update or replace users set email = 'b@example.com' where id = 1"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Permission denied: need delete-row on data/users"
+    ]
+    assert (await db.execute("select id, email from users order by id")).dicts() == [
+        {"id": 1, "email": "a@example.com"},
+        {"id": 2, "email": "b@example.com"},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_execute_write_update_requires_insert_row_permission():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    },
+                    "tables": {
+                        "users": {
+                            "permissions": {
+                                "update-row": {"id": "writer"},
+                                "delete-row": {"id": "writer"},
+                                "view-table": {"id": "writer"},
+                            }
+                        }
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_update_requires_insert", name="data")
+    await db.execute_write("create table users (id integer primary key, name text)")
+    await db.execute_write("insert into users (id, name) values (1, 'Alice')")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={"sql": "update users set name = 'Alicia' where id = 1"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Permission denied: need insert-row on data/users"
+    ]
+    assert (await db.execute("select name from users where id = 1")).first()[0] == "Alice"
+
+
 @pytest.mark.asyncio
 async def test_execute_write_insert_select_requires_view_table_on_source():
     ds = Datasette(
@@ -1659,7 +1845,13 @@ async def test_execute_write_insert_select_requires_view_table_on_source():
                         "secret": {
                             "permissions": {"view-table": {"id": "someone-else"}}
                         },
-                        "public_log": {"permissions": {"insert-row": {"id": "writer"}}},
+                        "public_log": {
+                            "permissions": {
+                                "insert-row": {"id": "writer"},
+                                "update-row": {"id": "writer"},
+                                "delete-row": {"id": "writer"},
+                            }
+                        },
                     },
                 }
             }
@@ -1740,6 +1932,8 @@ async def test_execute_write_rejects_function_operations():
                         "dogs": {
                             "permissions": {
                                 "insert-row": {"id": "writer"},
+                                "update-row": {"id": "writer"},
+                                "delete-row": {"id": "writer"},
                             }
                         }
                     },
@@ -2117,6 +2311,8 @@ async def test_user_writable_query_execution_rechecks_table_permissions():
                         "dogs": {
                             "permissions": {
                                 "insert-row": {"id": "alice"},
+                                "update-row": {"id": "alice"},
+                                "delete-row": {"id": "alice"},
                             }
                         }
                     },

From 1932f8429fd3259d48fb848fdf893f9a004276e9 Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Wed, 27 May 2026 16:14:50 -0700
Subject: [PATCH 5/8] Deny user-authored schema table reads in write SQL

Stop marking sqlite_master and sqlite_schema reads as internal as soon as the SQLite authorizer reports them. The later DDL-aware pass still treats schema catalog access as internal when it accompanies semantic CREATE, ALTER, or DROP operations.

This makes explicit catalog reads in write SQL fall through to the deny-by-default path as unsupported read schema operations, preventing queries from copying private table definitions into writable tables.

Refs https://github.com/simonw/datasette/pull/2749#issuecomment-4559073803
---
 datasette/utils/sql_analysis.py  |  1 -
 datasette/views/query_helpers.py |  4 +-
 tests/test_queries.py            | 73 +++++++++++++++++++++++++++-----
 tests/test_utils_sql_analysis.py | 20 +++++++++
 4 files changed, 84 insertions(+), 14 deletions(-)

diff --git a/datasette/utils/sql_analysis.py b/datasette/utils/sql_analysis.py
index 8963da77..91216501 100644
--- a/datasette/utils/sql_analysis.py
+++ b/datasette/utils/sql_analysis.py
@@ -256,7 +256,6 @@ def analyze_sql_tables(
                 target=arg1,
                 source=source,
                 column=column,
-                internal=target_type == "schema",
             )
             return sqlite3.SQLITE_OK
 
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index 7f3ef1bc..0e3d4e01 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -217,9 +217,7 @@ def _analysis_rows(analysis: SQLAnalysis) -> list[dict[str, object]]:
     rows = []
     for operation in _display_operations(analysis):
         permissions = permission_requirements_for_operation(operation)
-        required_permission = ", ".join(
-            permission.action for permission in permissions
-        )
+        required_permission = ", ".join(permission.action for permission in permissions)
         rows.append(
             {
                 "operation": operation.operation,
diff --git a/tests/test_queries.py b/tests/test_queries.py
index fcd19d1c..40bc5052 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -1643,9 +1643,9 @@ async def test_execute_write_post_requires_database_and_table_permissions():
         "Permission denied: need update-row on data/dogs"
     ]
 
-    ds.config["databases"]["data"]["tables"]["dogs"]["permissions"][
-        "update-row"
-    ] = {"id": "writer"}
+    ds.config["databases"]["data"]["tables"]["dogs"]["permissions"]["update-row"] = {
+        "id": "writer"
+    }
     missing_delete_permission = await ds.client.post(
         "/data/-/execute-write",
         actor={"id": "writer"},
@@ -1660,9 +1660,9 @@ async def test_execute_write_post_requires_database_and_table_permissions():
         "Permission denied: need delete-row on data/dogs"
     ]
 
-    ds.config["databases"]["data"]["tables"]["dogs"]["permissions"][
-        "delete-row"
-    ] = {"id": "writer"}
+    ds.config["databases"]["data"]["tables"]["dogs"]["permissions"]["delete-row"] = {
+        "id": "writer"
+    }
     allowed = await ds.client.post(
         "/data/-/execute-write",
         actor={"id": "writer"},
@@ -1719,8 +1719,7 @@ async def test_execute_write_insert_or_replace_requires_delete_row_permission():
         actor={"id": "writer"},
         json={
             "sql": (
-                "insert or replace into users(id, email) "
-                "values (3, 'b@example.com')"
+                "insert or replace into users(id, email) " "values (3, 'b@example.com')"
             )
         },
     )
@@ -1773,7 +1772,9 @@ async def test_execute_write_update_or_replace_requires_delete_row_permission():
     denied_response = await ds.client.post(
         "/data/-/execute-write",
         actor={"id": "writer"},
-        json={"sql": "update or replace users set email = 'b@example.com' where id = 1"},
+        json={
+            "sql": "update or replace users set email = 'b@example.com' where id = 1"
+        },
     )
 
     assert denied_response.status_code == 403
@@ -1826,7 +1827,9 @@ async def test_execute_write_update_requires_insert_row_permission():
     assert denied_response.json()["errors"] == [
         "Permission denied: need insert-row on data/users"
     ]
-    assert (await db.execute("select name from users where id = 1")).first()[0] == "Alice"
+    assert (await db.execute("select name from users where id = 1")).first()[
+        0
+    ] == "Alice"
 
 
 @pytest.mark.asyncio
@@ -1876,6 +1879,56 @@ async def test_execute_write_insert_select_requires_view_table_on_source():
     assert (await db.execute("select value from public_log")).dicts() == []
 
 
+@pytest.mark.asyncio
+async def test_execute_write_rejects_sqlite_master_reads():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    },
+                    "tables": {
+                        "secret": {
+                            "permissions": {"view-table": {"id": "someone-else"}}
+                        },
+                        "log": {
+                            "permissions": {
+                                "insert-row": {"id": "writer"},
+                                "update-row": {"id": "writer"},
+                                "delete-row": {"id": "writer"},
+                            }
+                        },
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("execute_write_sqlite_master_read", name="data")
+    await db.execute_write("create table secret (value text)")
+    await db.execute_write("create table log (value text)")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={
+            "sql": (
+                "insert into log " "select sql from sqlite_master where name = 'secret'"
+            )
+        },
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Unsupported SQL operation: read schema"
+    ]
+    assert (await db.execute("select value from log")).dicts() == []
+
+
 @pytest.mark.asyncio
 async def test_execute_write_create_table_as_select_requires_view_table_on_source():
     ds = Datasette(
diff --git a/tests/test_utils_sql_analysis.py b/tests/test_utils_sql_analysis.py
index 2ae11502..f931be51 100644
--- a/tests/test_utils_sql_analysis.py
+++ b/tests/test_utils_sql_analysis.py
@@ -65,6 +65,26 @@ def test_analyze_uses_sqlite_schema_as_default_database(conn):
     }
 
 
+def test_analyze_user_schema_table_read_is_not_internal(conn):
+    analysis = analyze_sql_tables(
+        conn,
+        "insert into log select sql from sqlite_master where name = 'dogs'",
+        database_name="data",
+    )
+
+    assert {
+        "operation": "read",
+        "target_type": "schema",
+        "database": "data",
+        "sqlite_schema": "main",
+        "table": None,
+        "target": "sqlite_master",
+        "columns": ("name", "sql"),
+        "source": None,
+        "internal": False,
+    } in [operation_dict(operation) for operation in analysis.operations]
+
+
 def operation_dict(operation):
     return {
         "operation": operation.operation,

From 951f5a9f306ebe0bb8b3668ee698dc6cb6051d78 Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Wed, 27 May 2026 16:30:05 -0700
Subject: [PATCH 6/8] Detect VACUUM in SQL analysis

Refs https://github.com/simonw/datasette/pull/2749#issuecomment-4559073803
---
 datasette/stored_queries.py      |  1 +
 datasette/utils/sql_analysis.py  | 33 +++++++++++++++++++++++-
 tests/test_queries.py            | 31 ++++++++++++++++++++++
 tests/test_utils_sql_analysis.py | 44 ++++++++++++++++++++++++++++++++
 4 files changed, 108 insertions(+), 1 deletion(-)

diff --git a/datasette/stored_queries.py b/datasette/stored_queries.py
index cf44a9ff..6746124a 100644
--- a/datasette/stored_queries.py
+++ b/datasette/stored_queries.py
@@ -720,6 +720,7 @@ def operation_is_write(operation: Operation) -> bool:
         "pragma",
         "analyze",
         "reindex",
+        "vacuum",
         "unknown",
     }
 
diff --git a/datasette/utils/sql_analysis.py b/datasette/utils/sql_analysis.py
index 91216501..f2eb903f 100644
--- a/datasette/utils/sql_analysis.py
+++ b/datasette/utils/sql_analysis.py
@@ -22,6 +22,7 @@ SQLOperation = Literal[
     "pragma",
     "analyze",
     "reindex",
+    "vacuum",
     "unknown",
 ]
 SQLTargetType = Literal[
@@ -423,10 +424,40 @@ def analyze_sql_tables(
 
     conn.set_authorizer(authorizer)
     try:
-        conn.execute("EXPLAIN " + sql, params if params is not None else {}).fetchall()
+        explain_rows = conn.execute(
+            "EXPLAIN " + sql, params if params is not None else {}
+        ).fetchall()
     finally:
         conn.set_authorizer(None)
 
+    if not operations:
+        vacuum_row = next((row for row in explain_rows if row[1] == "Vacuum"), None)
+        if vacuum_row is not None:
+            schema_by_index = {
+                row[0]: row[1] for row in conn.execute("PRAGMA database_list")
+            }
+            sqlite_schema = schema_by_index.get(vacuum_row[2])
+            database = database_for_schema(sqlite_schema)
+            record(
+                "vacuum",
+                "database",
+                database=database,
+                table=None,
+                sqlite_schema=sqlite_schema,
+                target=database,
+                source=None,
+            )
+        else:
+            record(
+                "unknown",
+                "statement",
+                database=database_name,
+                table=None,
+                sqlite_schema=None,
+                target=None,
+                source=None,
+            )
+
     has_schema_operation = any(
         key.target_type in {"table", "index", "view", "trigger", "virtual-table"}
         and key.operation in {"create", "alter", "drop"}
diff --git a/tests/test_queries.py b/tests/test_queries.py
index 40bc5052..bf371a80 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -2011,6 +2011,37 @@ async def test_execute_write_rejects_function_operations():
     assert (await db.execute("select name from dogs")).dicts() == []
 
 
+@pytest.mark.asyncio
+async def test_execute_write_rejects_vacuum_operation():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    }
+                }
+            }
+        },
+    )
+    ds.add_memory_database("execute_write_vacuum_operation", name="data")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        json={"sql": "vacuum"},
+    )
+
+    assert denied_response.status_code == 403
+    assert denied_response.json()["errors"] == [
+        "Unsupported SQL operation: vacuum database"
+    ]
+
+
 @pytest.mark.asyncio
 async def test_execute_write_create_table_uses_create_table_permission():
     ds = Datasette(
diff --git a/tests/test_utils_sql_analysis.py b/tests/test_utils_sql_analysis.py
index f931be51..df4b3625 100644
--- a/tests/test_utils_sql_analysis.py
+++ b/tests/test_utils_sql_analysis.py
@@ -129,6 +129,50 @@ def test_analyze_create_table_operation():
     ]
 
 
+def test_analyze_vacuum_operation():
+    conn = sqlite3.connect(":memory:")
+    try:
+        analysis = analyze_sql_tables(conn, "vacuum", database_name="data")
+    finally:
+        conn.close()
+
+    assert [operation_dict(operation) for operation in analysis.operations] == [
+        {
+            "operation": "vacuum",
+            "target_type": "database",
+            "database": "data",
+            "sqlite_schema": "main",
+            "table": None,
+            "target": "data",
+            "columns": (),
+            "source": None,
+            "internal": False,
+        }
+    ]
+
+
+def test_analyze_statement_with_no_authorizer_callbacks_is_unknown():
+    conn = sqlite3.connect(":memory:")
+    try:
+        analysis = analyze_sql_tables(conn, "reindex", database_name="data")
+    finally:
+        conn.close()
+
+    assert [operation_dict(operation) for operation in analysis.operations] == [
+        {
+            "operation": "unknown",
+            "target_type": "statement",
+            "database": "data",
+            "sqlite_schema": None,
+            "table": None,
+            "target": None,
+            "columns": (),
+            "source": None,
+            "internal": False,
+        }
+    ]
+
+
 def test_analyze_transaction_operation(conn):
     analysis = analyze_sql_tables(conn, "commit", database_name="data")
 

From 11bddc891918849e7c4a006c64d0217072aa499c Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Wed, 27 May 2026 16:51:12 -0700
Subject: [PATCH 7/8] Deny VACUUM in user-authored SQL

Reject VACUUM explicitly during write-query permission analysis so arbitrary write SQL and untrusted stored write queries cannot run it, even when the actor has execute-write-sql.

Refs https://github.com/simonw/datasette/pull/2749#issuecomment-4559073803 (P3)
---
 datasette/stored_queries.py      |  16 ++++
 datasette/views/database.py      |  23 ++++-
 datasette/views/execute_write.py |   6 +-
 datasette/views/query_helpers.py |   9 +-
 tests/test_queries.py            | 153 ++++++++++++++++++++++++++++++-
 5 files changed, 199 insertions(+), 8 deletions(-)

diff --git a/datasette/stored_queries.py b/datasette/stored_queries.py
index 6746124a..fd1cabf3 100644
--- a/datasette/stored_queries.py
+++ b/datasette/stored_queries.py
@@ -15,6 +15,13 @@ if TYPE_CHECKING:
 
 UNCHANGED = object()
 
+
+class QueryWriteRejected(Exception):
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(message)
+
+
 QUERY_OPTION_FIELDS = (
     "hide_sql",
     "fragment",
@@ -703,6 +710,12 @@ def operation_should_be_ignored(operation: Operation) -> bool:
     return operation.internal or operation.operation == "select"
 
 
+def operation_forbidden_message(operation: Operation) -> str | None:
+    if operation.operation == "vacuum":
+        return "VACUUM is not allowed in user-supplied SQL"
+    return None
+
+
 def operation_is_write(operation: Operation) -> bool:
     return operation.operation in {
         "insert",
@@ -746,6 +759,9 @@ async def ensure_query_write_permissions(
     for operation in analysis.operations:
         if operation_should_be_ignored(operation):
             continue
+        forbidden_message = operation_forbidden_message(operation)
+        if forbidden_message is not None:
+            raise QueryWriteRejected(forbidden_message)
         permissions = permission_requirements_for_operation(operation)
         if not permissions:
             raise Forbidden(
diff --git a/datasette/views/database.py b/datasette/views/database.py
index b558b002..ae1cf375 100644
--- a/datasette/views/database.py
+++ b/datasette/views/database.py
@@ -13,7 +13,7 @@ import textwrap
 from datasette.events import AlterTableEvent, CreateTableEvent, InsertRowsEvent
 from datasette.database import QueryInterrupted
 from datasette.resources import DatabaseResource, QueryResource
-from datasette.stored_queries import stored_query_to_dict
+from datasette.stored_queries import QueryWriteRejected, stored_query_to_dict
 from datasette.utils import (
     add_cors_headers,
     await_me_maybe,
@@ -453,9 +453,24 @@ class QueryView(View):
         ):
             raise Forbidden("You do not have permission to view this query")
 
-        await _ensure_stored_query_execution_permissions(
-            datasette, db, stored_query, request.actor
-        )
+        try:
+            await _ensure_stored_query_execution_permissions(
+                datasette, db, stored_query, request.actor
+            )
+        except QueryWriteRejected as ex:
+            if request.headers.get("accept") == "application/json" or request.args.get(
+                "_json"
+            ):
+                return Response.json(
+                    {
+                        "ok": False,
+                        "message": ex.message,
+                        "redirect": None,
+                    },
+                    status=403,
+                )
+            datasette.add_message(request, ex.message, datasette.ERROR)
+            return Response.redirect(stored_query.on_error_redirect or request.path)
 
         # If database is immutable, return an error
         if not db.is_mutable:
diff --git a/datasette/views/execute_write.py b/datasette/views/execute_write.py
index 19006ac5..57c4d78e 100644
--- a/datasette/views/execute_write.py
+++ b/datasette/views/execute_write.py
@@ -163,13 +163,15 @@ class ExecuteWriteView(BaseView):
         except QueryValidationError as ex:
             if _wants_json(request, is_json, data):
                 return _block_framing(_error([ex.message], ex.status))
+            if ex.flash:
+                self.ds.add_message(request, ex.message, self.ds.ERROR)
             return await self._render_form(
                 request,
                 db,
                 sql=sql or "",
                 parameter_values=provided_params,
-                analysis_error=ex.message,
-                execution_message=ex.message,
+                analysis_error=None if ex.flash else ex.message,
+                execution_message=None if ex.flash else ex.message,
                 execution_ok=False,
                 status=ex.status,
             )
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index 0e3d4e01..92328ff3 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -3,6 +3,7 @@ import re
 
 from datasette.resources import DatabaseResource
 from datasette.stored_queries import (
+    QueryWriteRejected,
     StoredQuery,
     operation_is_write,
     operation_should_be_ignored,
@@ -47,9 +48,11 @@ _query_write_fields = {
 
 
 class QueryValidationError(Exception):
-    def __init__(self, message, status=400):
+    def __init__(self, message, status=400, *, flash=False):
         self.message = message
         self.status = status
+        self.flash = flash
+        super().__init__(message)
 
 
 def _actor_id(actor):
@@ -194,6 +197,8 @@ async def _analyze_user_query(datasette, db, sql, *, actor):
             await datasette.ensure_query_write_permissions(
                 db.name, sql, actor=actor, analysis=analysis
             )
+        except QueryWriteRejected as ex:
+            raise QueryValidationError(ex.message, status=403, flash=True) from ex
         except Forbidden as ex:
             raise QueryValidationError(str(ex), status=403) from ex
     else:
@@ -297,6 +302,8 @@ async def _prepare_execute_write(datasette, db, sql, params, actor):
         await datasette.ensure_query_write_permissions(
             db.name, sql, actor=actor, analysis=analysis
         )
+    except QueryWriteRejected as ex:
+        raise QueryValidationError(ex.message, status=403, flash=True) from ex
     except Forbidden as ex:
         raise QueryValidationError(str(ex), status=403) from ex
     return parameter_names, params, analysis
diff --git a/tests/test_queries.py b/tests/test_queries.py
index bf371a80..b6e1637d 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -2038,10 +2038,161 @@ async def test_execute_write_rejects_vacuum_operation():
 
     assert denied_response.status_code == 403
     assert denied_response.json()["errors"] == [
-        "Unsupported SQL operation: vacuum database"
+        "VACUUM is not allowed in user-supplied SQL"
     ]
 
 
+@pytest.mark.asyncio
+async def test_execute_write_form_rejects_vacuum_operation_with_flash_error():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    }
+                }
+            }
+        },
+    )
+    ds.add_memory_database("execute_write_vacuum_operation_form", name="data")
+    await ds.invoke_startup()
+
+    denied_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "writer"},
+        data={"sql": "vacuum"},
+    )
+
+    assert denied_response.status_code == 403
+    assert (
+        '<p class="message-error">VACUUM is not allowed in user-supplied SQL</p>'
+        in denied_response.text
+    )
+    assert denied_response.text.count("VACUUM is not allowed in user-supplied SQL") == 1
+
+
+@pytest.mark.asyncio
+async def test_untrusted_stored_write_query_rejects_vacuum_operation():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "view-query": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    }
+                }
+            }
+        },
+    )
+    ds.add_memory_database("stored_query_vacuum_operation", name="data")
+    await ds.invoke_startup()
+    await ds.add_query(
+        "data",
+        "vacuum_db",
+        "vacuum",
+        is_write=True,
+        is_trusted=False,
+        source="user",
+        owner_id="writer",
+    )
+
+    denied_response = await ds.client.post(
+        "/data/vacuum_db?_json=1",
+        actor={"id": "writer"},
+        data={},
+    )
+
+    assert denied_response.status_code == 403
+    assert "VACUUM is not allowed in user-supplied SQL" in denied_response.text
+
+
+@pytest.mark.asyncio
+async def test_untrusted_stored_write_query_rejects_vacuum_operation_with_flash_error():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "view-query": {"id": "writer"},
+                        "execute-write-sql": {"id": "writer"},
+                    }
+                }
+            }
+        },
+    )
+    ds.add_memory_database("stored_query_vacuum_operation_form", name="data")
+    await ds.invoke_startup()
+    await ds.add_query(
+        "data",
+        "vacuum_db",
+        "vacuum",
+        is_write=True,
+        is_trusted=False,
+        source="user",
+        owner_id="writer",
+    )
+
+    denied_response = await ds.client.post(
+        "/data/vacuum_db",
+        actor={"id": "writer"},
+        data={},
+    )
+
+    assert denied_response.status_code == 302
+    assert denied_response.headers["location"] == "/data/vacuum_db"
+    assert ds.unsign(denied_response.cookies["ds_messages"], "messages") == [
+        ["VACUUM is not allowed in user-supplied SQL", ds.ERROR]
+    ]
+
+
+@pytest.mark.asyncio
+async def test_trusted_stored_write_query_skips_vacuum_filtering():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "view-database": {"id": "writer"},
+                        "view-query": {"id": "writer"},
+                    }
+                }
+            }
+        },
+    )
+    ds.add_memory_database("trusted_stored_query_vacuum", name="data")
+    await ds.invoke_startup()
+    await ds.add_query(
+        "data",
+        "trusted_vacuum",
+        "vacuum",
+        is_write=True,
+        is_trusted=True,
+        source="config",
+    )
+
+    response = await ds.client.post(
+        "/data/trusted_vacuum?_json=1",
+        actor={"id": "writer"},
+        data={},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["ok"] is True
+
+
 @pytest.mark.asyncio
 async def test_execute_write_create_table_uses_create_table_permission():
     ds = Datasette(

From 0c5053cdf64a0dc2d1e9808fa712b88233760512 Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Wed, 27 May 2026 17:26:50 -0700
Subject: [PATCH 8/8] Docs for /<database>/-/execute-write JSON API

Closes #2750, refs #2742
---
 docs/json_api.rst | 62 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 62 insertions(+)

diff --git a/docs/json_api.rst b/docs/json_api.rst
index 48c70af6..fffc16d7 100644
--- a/docs/json_api.rst
+++ b/docs/json_api.rst
@@ -505,6 +505,68 @@ The JSON write API
 
 Datasette provides a write API for JSON data. This is a POST-only API that requires an authenticated API token, see :ref:`CreateTokenView`. The token will need to have the specified :ref:`authentication_permissions`.
 
+.. _ExecuteWriteView:
+
+Executing write SQL
+~~~~~~~~~~~~~~~~~~~
+
+Actors with the :ref:`actions_execute_write_sql` permission can execute arbitrary writable SQL against a mutable database using ``/-/execute-write``.
+
+::
+
+    POST /<database>/-/execute-write
+    Content-Type: application/json
+    Authorization: Bearer dstok_<rest-of-token>
+
+The request body must include a ``"sql"`` string. Named SQL parameters can be provided using the optional ``"params"`` object:
+
+.. code-block:: json
+
+    {
+        "sql": "insert into dogs (name) values (:name)",
+        "params": {
+            "name": "Cleo"
+        }
+    }
+
+The SQL must be writable. Read-only ``select`` queries should use the regular :ref:`custom SQL query API <sql>` instead.
+
+Datasette analyzes the SQL before executing it. The actor must have ``execute-write-sql`` permission for the database, and must also have any permissions required by the operations in the SQL. For example, inserts and updates against a table require ``insert-row``, ``update-row`` and ``delete-row`` permissions for that table. Reads performed as part of the write, such as ``insert into dogs select ... from other_table``, require ``view-table`` permission on the source table.
+
+A successful response includes a message, the SQLite ``rowcount`` and a summary of the operations that were executed:
+
+The shape of the ``"analysis"`` block is not yet considered a stable API and may change in future Datasette releases.
+
+.. code-block:: json
+
+    {
+        "ok": true,
+        "message": "Query executed, 1 row affected",
+        "rowcount": 1,
+        "analysis": [
+            {
+                "operation": "insert",
+                "database": "data",
+                "table": "dogs",
+                "required_permission": "insert-row, update-row, delete-row",
+                "source": null
+            }
+        ]
+    }
+
+If SQLite reports ``-1`` for the row count, the message will be ``"Query executed"``.
+
+Errors use the standard Datasette error format:
+
+.. code-block:: json
+
+    {
+        "ok": false,
+        "errors": [
+            "Permission denied: need execute-write-sql"
+        ]
+    }
+
 .. _TableInsertView:
 
 Inserting rows