execute-write: Check view-table for every table in a CREATE VIEW

Refs GHSA-53fc-rhfg-h7qp

Co-authored-by: Alex Garcia <15178711+asg017@users.noreply.github.com>

datasette/utils/sql_analysis.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
 from typing import Literal
 
+from datasette.utils import escape_sqlite
 from datasette.utils.sqlite import SQLiteTableType, sqlite3, sqlite_table_type
 
 SQLOperation = Literal[
@@ -208,7 +209,9 @@ def analyze_sql_tables(
 
     This function is synchronous and connection-based. It temporarily installs a
     SQLite authorizer, prepares ``EXPLAIN <sql>``, and returns the operation
-    callbacks observed while SQLite compiles the statement.
+    callbacks observed while SQLite compiles the statement. ``CREATE VIEW`` is
+    additionally executed inside a rolled-back savepoint so its source-table reads
+    can be discovered by analyzing a query against the temporary view.
     """
     operations: dict[OperationKey, set[str]] = {}
 
@@ -532,7 +535,7 @@ def analyze_sql_tables(
             return None
         return table_kind_cache[(key.sqlite_schema, key.table)]
 
-    return SQLAnalysis(
+    analysis = SQLAnalysis(
         operations=tuple(
             Operation(
                 operation=key.operation,
@@ -549,3 +552,58 @@ def analyze_sql_tables(
             for key, columns in operations.items()
         )
     )
+
+    # SQLite does not resolve the SELECT body of a view when preparing CREATE
+    # VIEW, so its authorizer does not report reads from the view's source
+    # tables. Temporarily create the view, analyze a query against it (which
+    # does resolve the body), then roll the schema change back. Database-level
+    # callers use an isolated writable connection for this analysis.
+    create_view_operations = tuple(
+        operation
+        for operation in analysis.operations
+        if operation.operation == "create" and operation.target_type == "view"
+    )
+    if not create_view_operations:
+        return analysis
+
+    savepoint = "datasette_analyze_create_view"
+    conn.execute(f"SAVEPOINT {savepoint}")
+    try:
+        conn.execute(sql, params if params is not None else {})
+        dependency_reads = []
+        for view_operation in create_view_operations:
+            if view_operation.sqlite_schema is None or view_operation.table is None:
+                raise sqlite3.OperationalError(
+                    "Could not determine the created view name"
+                )
+            quoted_schema = escape_sqlite(view_operation.sqlite_schema)
+            quoted_view = escape_sqlite(view_operation.table)
+            qualified_view = f"{quoted_schema}.{quoted_view}"
+            view_analysis = analyze_sql_tables(
+                conn,
+                f"SELECT * FROM {qualified_view}",
+                database_name=database_name,
+                schema_to_database=schema_to_database,
+            )
+            dependency_reads.extend(
+                operation
+                for operation in view_analysis.operations
+                if operation.operation == "read"
+                and not (
+                    operation.sqlite_schema == view_operation.sqlite_schema
+                    and operation.table == view_operation.table
+                )
+            )
+    finally:
+        conn.execute(f"ROLLBACK TO {savepoint}")
+        conn.execute(f"RELEASE {savepoint}")
+
+    existing_operations = set(analysis.operations)
+    return SQLAnalysis(
+        operations=analysis.operations
+        + tuple(
+            operation
+            for operation in dependency_reads
+            if operation not in existing_operations
+        )
+    )

tests/test_queries.py

@@ -3248,74 +3248,6 @@ async def test_execute_write_create_table_uses_create_table_permission():
     assert not await db.table_exists("should_not_exist")
 
 
-@pytest.mark.asyncio
-async def test_execute_write_create_view_uses_create_view_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-view": {"id": "creator"},
-                    }
-                }
-            },
-        },
-    )
-    db = ds.add_memory_database("execute_write_create_view", name="data")
-    await db.execute_write("create table dogs (id integer primary key, name text)")
-    await ds.invoke_startup()
-
-    analysis_response = await ds.client.get(
-        "/data/-/execute-write/analyze",
-        actor={"id": "creator"},
-        params={"sql": "create view dog_names as select id, name from dogs"},
-    )
-    allowed_response = await ds.client.post(
-        "/data/-/execute-write",
-        actor={"id": "creator"},
-        json={"sql": "create view dog_names as select id, name from dogs"},
-    )
-    row_permission_response = await ds.client.post(
-        "/data/-/execute-write",
-        actor={"id": "row-writer"},
-        json={"sql": "create view should_not_exist as select id from dogs"},
-    )
-
-    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": "dog_names",
-            "required_permission": "create-view",
-            "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.view_exists("dog_names")
-
-    assert row_permission_response.status_code == 403
-    assert row_permission_response.json()["errors"] == [
-        "Permission denied: need create-view on data"
-    ]
-    assert not await db.view_exists("should_not_exist")
-
-
 @pytest.mark.parametrize(
     (
         "database_name",