Docs for /<database>/-/execute-write JSON API

Closes #2750, refs #2742

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

datasette/stored_queries.py

@@ -2,14 +2,26 @@ 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()
 
+
+class QueryWriteRejected(Exception):
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(message)
+
+
 QUERY_OPTION_FIELDS = (
     "hide_sql",
     "fragment",
@@ -583,20 +595,158 @@ async def list_queries(
     )
 
 
+@dataclass(frozen=True)
+class PermissionRequirement:
+    action: str
+    resource: Resource
+
+
+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"
+        and operation.database is not None
+        and operation.table is not None
+    ):
+        return (
+            PermissionRequirement(
+                action="view-table",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
+        )
+    if (
+        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 (
+            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 ()
+        return (
+            PermissionRequirement(
+                action="create-table",
+                resource=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 (
+            PermissionRequirement(
+                action="alter-table",
+                resource=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 (
+            PermissionRequirement(
+                action="drop-table",
+                resource=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 (
+            PermissionRequirement(
+                action="alter-table",
+                resource=TableResource(
+                    database=operation.database, table=operation.table
+                ),
+            ),
+        )
+    return ()
+
+
+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",
+        "update",
+        "delete",
+        "create",
+        "alter",
+        "drop",
+        "begin",
+        "commit",
+        "rollback",
+        "savepoint",
+        "attach",
+        "detach",
+        "pragma",
+        "analyze",
+        "reindex",
+        "vacuum",
+        "unknown",
+    }
+
+
 async def ensure_query_write_permissions(
-    datasette: Any,
+    datasette: Datasette,
     database: str,
     sql: str,
     *,
-    actor: dict[str, Any] | None = None,
-    params: dict[str, Any] | None = None,
-    analysis: Any = None,
-) -> Any:
-    write_actions = {
-        "insert": "insert-row",
-        "update": "update-row",
-        "delete": "delete-row",
-    }
+    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 +756,29 @@ 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:
+    for operation in analysis.operations:
+        if operation_should_be_ignored(operation):
             continue
-        if access.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),
-            actor=actor,
-        ):
+        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(
-                f"Permission denied: need {action} on {access.database}/{access.table}"
+                "Unsupported SQL operation: {} {}".format(
+                    operation.operation, operation.target_type
+                )
             )
+        if operation.database != database:
+            raise Forbidden("Writable queries may not access attached databases")
+        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

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 %}>

datasette/utils/sql_analysis.py

@@ -3,22 +3,76 @@ from typing import Literal
 
 from datasette.utils.sqlite import sqlite3
 
+SQLOperation = Literal[
+    "read",
+    "insert",
+    "update",
+    "delete",
+    "select",
+    "function",
+    "create",
+    "alter",
+    "drop",
+    "begin",
+    "commit",
+    "rollback",
+    "savepoint",
+    "attach",
+    "detach",
+    "pragma",
+    "analyze",
+    "reindex",
+    "vacuum",
+    "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)
-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 +82,114 @@ _ACTION_TO_OPERATION: dict[int, SQLTableOperation] = {
     sqlite3.SQLITE_DELETE: "delete",
 }
 
+# Values are (operation, target_type) pairs used to construct Operation objects.
+_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: 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"),
+}
+
+
+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"),
+    ("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"),
+)
+for schema_action in _TEMP_SCHEMA_ACTIONS:
+    _add_schema_action(*schema_action)
+
+_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(
     conn,
@@ -38,15 +200,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,45 +215,304 @@ 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,
+            )
             return sqlite3.SQLITE_OK
 
-        key = (
-            operation,
-            database_for_schema(sqlite_schema),
-            arg1,
-            sqlite_schema,
-            source,
+        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
+
+        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,
         )
-        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)
     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)
 
-    return SQLAnalysis(
-        table_accesses=tuple(
-            SQLTableAccess(
-                operation=operation,
+    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=table,
+                table=None,
                 sqlite_schema=sqlite_schema,
-                columns=tuple(sorted(columns)),
-                source=source,
+                target=database,
+                source=None,
             )
-            for (
-                operation,
-                database,
-                table,
-                sqlite_schema,
-                source,
-            ), columns in accesses.items()
+        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"}
+        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(
+            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=key.source,
+                internal=operation_is_internal(key),
+            )
+            for key, columns in operations.items()
         )
     )

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:

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,
@@ -165,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,
             )
@@ -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(

datasette/views/query_helpers.py

@@ -1,8 +1,14 @@
 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 (
+    QueryWriteRejected,
+    StoredQuery,
+    operation_is_write,
+    operation_should_be_ignored,
+    permission_requirements_for_operation,
+)
 from datasette.utils import (
     named_parameters as derive_named_parameters,
     escape_sqlite,
@@ -12,6 +18,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]+$")
 
@@ -41,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):
@@ -123,11 +132,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):
@@ -191,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:
@@ -201,34 +209,51 @@ 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 _display_operations(analysis: SQLAnalysis) -> list[Operation]:
+    operations = []
+    for operation in analysis.operations:
+        if operation_should_be_ignored(operation):
+            continue
+        operations.append(operation)
+    return operations
 
 
-async def _analysis_rows_with_permissions(datasette, analysis, actor):
+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)
+        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"]
-        if permission:
-            row["allowed"] = await datasette.allowed(
-                action=permission,
-                resource=TableResource(row["database"], row["table"]),
-                actor=actor,
-            )
+    is_write = _analysis_is_write(analysis)
+    for row, operation in zip(rows, _display_operations(analysis)):
+        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:
             row["allowed"] = None
     return rows
@@ -277,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
@@ -326,7 +353,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
@@ -340,6 +367,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)
@@ -356,9 +384,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
@@ -398,15 +424,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 = (

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

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):

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),

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"},
                         }
                     }
                 }
@@ -1181,11 +1183,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 +1256,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 +1376,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
@@ -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,687 @@ 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(
+        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"},
+                                "update-row": {"id": "writer"},
+                                "delete-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_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(
+        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"},
+                                "update-row": {"id": "writer"},
+                                "delete-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_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"] == [
+        "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(
+        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"},
+                                "view-table": {"id": "alterer"},
+                            }
+                        }
+                    },
+                }
+            },
+        },
+    )
+    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)
@@ -1829,6 +2546,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"},
                             }
                         }
                     },

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,231 @@ 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 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,
+        "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_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")
+
+    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_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,
@@ -70,7 +293,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 +310,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 +324,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 +344,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 +358,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 +386,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),
     }