Disallow update/delete of private queries

If a user does not own a private query they cannot update
or delete it either, even if they have global update-query.

https://github.com/simonw/datasette/pull/2741/changes#r3306417463

datasette/default_permissions/defaults.py

@@ -77,36 +77,31 @@ async def default_query_permissions_sql(
 ) -> Optional[PermissionSQL]:
     actor_id = actor.get("id") if isinstance(actor, dict) else None
 
-    if action in {"update-query", "delete-query"}:
-        if actor_id is None:
-            return None
-        # Query owner can update/delete query
-        return PermissionSQL(
-            sql="""
-            SELECT database_name AS parent, name AS child, 1 AS allow,
-              'query owner' AS reason
-            FROM queries
-            WHERE source = 'user'
-              AND owner_id = :query_owner_id
-            """,
-            params={"query_owner_id": actor_id},
-        )
-
-    if action != "view-query":
+    if action not in {"view-query", "update-query", "delete-query"}:
         return None
 
     params = {"query_owner_id": actor_id}
     rule_sqls = []
     if actor_id is not None:
-        # Query owner can view-query
-        rule_sqls.append("""
+        if action in {"update-query", "delete-query"}:
+            # Query owner can update/delete query
+            rule_sqls.append("""
+            SELECT database_name AS parent, name AS child, 1 AS allow,
+              'query owner' AS reason
+            FROM queries
+            WHERE source = 'user'
+              AND owner_id = :query_owner_id
+            """)
+        else:
+            # Query owner can view-query
+            rule_sqls.append("""
             SELECT database_name AS parent, name AS child, 1 AS allow,
               'query owner' AS reason
             FROM queries
             WHERE owner_id = :query_owner_id
             """)
 
-    # restriction_sql enforces private queries ONLY visible to owner
+    # restriction_sql enforces private queries ONLY visible/mutable by owner
     return PermissionSQL(
         sql="\nUNION ALL\n".join(rule_sqls) if rule_sqls else None,
         restriction_sql="""

tests/test_queries.py

@@ -1581,6 +1581,87 @@ async def test_query_owner_gets_update_delete_and_writable_view_defaults():
         )
 
 
+@pytest.mark.asyncio
+async def test_private_query_restricts_broad_update_delete_permissions():
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "update-query": {"id": "bob"},
+                        "delete-query": {"id": "bob"},
+                    },
+                },
+            },
+        },
+    )
+    ds.add_memory_database("query_broad_update_delete", name="data")
+    await ds.invoke_startup()
+    await ds.add_query(
+        "data",
+        "alice_private",
+        "select 1",
+        is_private=True,
+        source="user",
+        owner_id="alice",
+    )
+    await ds.add_query(
+        "data",
+        "alice_public",
+        "select 2",
+        is_private=False,
+        source="user",
+        owner_id="alice",
+    )
+
+    for action in ("update-query", "delete-query"):
+        assert await ds.allowed(
+            action=action,
+            resource=QueryResource("data", "alice_private"),
+            actor={"id": "alice"},
+        )
+        assert not await ds.allowed(
+            action=action,
+            resource=QueryResource("data", "alice_private"),
+            actor={"id": "bob"},
+        )
+        assert await ds.allowed(
+            action=action,
+            resource=QueryResource("data", "alice_public"),
+            actor={"id": "bob"},
+        )
+
+    private_update_response = await ds.client.post(
+        "/data/alice_private/-/update",
+        actor={"id": "bob"},
+        json={"update": {"title": "Nope"}},
+    )
+    private_delete_response = await ds.client.post(
+        "/data/alice_private/-/delete",
+        actor={"id": "bob"},
+        json={},
+    )
+    public_update_response = await ds.client.post(
+        "/data/alice_public/-/update",
+        actor={"id": "bob"},
+        json={"update": {"title": "Bob can edit public queries"}},
+    )
+    public_delete_response = await ds.client.post(
+        "/data/alice_public/-/delete",
+        actor={"id": "bob"},
+        json={},
+    )
+
+    assert private_update_response.status_code == 403
+    assert private_delete_response.status_code == 403
+    assert public_update_response.status_code == 200
+    assert public_delete_response.status_code == 200
+    assert await ds.get_query("data", "alice_private") is not None
+    assert await ds.get_query("data", "alice_public") is None
+
+
 @pytest.mark.asyncio
 async def test_user_writable_query_execution_rechecks_table_permissions():
     ds = Datasette(