edit-schema-alter-table permission, refs #22

README.md

@@ -49,6 +49,7 @@ These permission checks will call the `permission_allowed()` plugin hook with th
 You can instead use more finely-grained permissions.
 
 - `edit-schema-create-table` allows users to create a new table. The `resource` will be the name of the database.
+- `edit-schema-alter-table` allows users to alter the schema of a table. The `resource` will be a tuple of `(database_name, table_name)`.
 
 ## Screenshot
 

datasette_edit_schema/__init__.py

@@ -29,9 +29,7 @@ def permission_allowed(actor, action, resource):
 @hookimpl
 def table_actions(datasette, actor, database, table):
     async def inner():
-        if not await datasette.permission_allowed(
-            actor, "edit-schema", resource=database, default=False
-        ):
+        if not await can_alter_table(datasette, actor, database, table):
             return []
         return [
             {
@@ -58,6 +56,19 @@ async def can_create_table(datasette, actor, database):
     return False
 
 
+async def can_alter_table(datasette, actor, database, table):
+    if await datasette.permission_allowed(
+        actor, "edit-schema", resource=database, default=False
+    ):
+        return True
+    # Or maybe they have edit-schema-create-table
+    if await datasette.permission_allowed(
+        actor, "edit-schema-alter-table", resource=(database, table), default=False
+    ):
+        return True
+    return False
+
+
 @hookimpl
 def database_actions(datasette, actor, database):
     async def inner():
@@ -262,7 +273,10 @@ async def edit_schema_table(request, datasette):
     table = unquote_plus(request.url_vars["table"])
     databases = get_databases(datasette)
     database_name = request.url_vars["database"]
-    await check_permissions(datasette, request, database_name)
+
+    if not await can_alter_table(datasette, request.actor, database_name, table):
+        raise Forbidden("Permission denied for edit-schema-alter-table")
+
     try:
         database = [db for db in databases if db.name == database_name][0]
     except IndexError:

tests/test_edit_schema.py

@@ -26,21 +26,43 @@ async def test_csrf_required(db_path):
 
 
 @pytest.mark.parametrize(
-    "authenticated,path,should_allow",
+    "actor_id,should_allow",
     (
-        (False, "/data/creatures", False),
-        (True, "/data/creatures", True),
+        (None, False),
+        ("user_with_edit_schema", True),
+        ("user_with_alter_table", True),
+        ("user_with_create_table", False),
+        ("user_with_no_perms", False),
     ),
 )
 @pytest.mark.asyncio
-async def test_table_actions(db_path, authenticated, path, should_allow):
-    ds = Datasette([db_path])
+async def test_table_actions(permission_plugin, ds, actor_id, should_allow):
+    ds._rules_allow = [
+        Rule(
+            actor_id="user_with_edit_schema",
+            action="edit-schema",
+            database="data",
+            resource=None,
+        ),
+        Rule(
+            actor_id="user_with_alter_table",
+            action="edit-schema-alter-table",
+            database="data",
+            resource="creatures",
+        ),
+        Rule(
+            actor_id="user_with_create_table",
+            action="edit-schema-create-table",
+            database="data",
+            resource=None,
+        ),
+    ]
     cookies = None
-    if authenticated:
-        cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
-    response = await ds.client.get(path, cookies=cookies)
+    if actor_id:
+        cookies = {"ds_actor": ds.sign({"a": {"id": actor_id}}, "actor")}
+    response = await ds.client.get("/data/creatures", cookies=cookies)
     assert response.status_code == 200
-    fragment = '<li><a href="/-/edit-schema{}">Edit table schema</a></li>'.format(path)
+    fragment = '<li><a href="/-/edit-schema/data/creatures">Edit table schema</a></li>'
     if should_allow:
         # Should have table action
         assert fragment in response.text
@@ -418,6 +440,79 @@ async def test_permission_create_table(permission_plugin, ds, rules_allow, shoul
     assert response.status_code == 302
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "rules_allow,should_work",
+    (
+        (
+            [
+                Rule(
+                    actor_id="user",
+                    action="edit-schema",
+                    database="data",
+                    resource=None,
+                ),
+            ],
+            True,
+        ),
+        (
+            [
+                Rule(
+                    actor_id="user2",
+                    action="edit-schema",
+                    database="data",
+                    resource=None,
+                ),
+            ],
+            False,
+        ),
+        (
+            [
+                Rule(
+                    actor_id="user",
+                    action="edit-schema-alter-table",
+                    database="data",
+                    resource="museums",
+                ),
+            ],
+            True,
+        ),
+        (
+            [
+                Rule(
+                    actor_id="user2",
+                    action="edit-schema-alter-table",
+                    database="data",
+                    resource="museums",
+                ),
+            ],
+            False,
+        ),
+    ),
+)
+async def test_permission_alter_table(permission_plugin, ds, rules_allow, should_work):
+    ds._rules_allow = rules_allow
+    cookies = {"ds_actor": ds.sign({"a": {"id": "user"}}, "actor")}
+    csrftoken_r = await ds.client.get("/-/edit-schema/data/museums", cookies=cookies)
+    if not should_work:
+        assert csrftoken_r.status_code == 403
+        return
+    assert csrftoken_r.status_code == 200
+    csrftoken = csrftoken_r.cookies["ds_csrftoken"]
+    cookies["ds_csrftoken"] = csrftoken
+    post_data = {
+        "action": "update_primary_key",
+        "primary_key": "name",
+        "csrftoken": csrftoken,
+    }
+    response = await ds.client.post(
+        "/-/edit-schema/data/museums",
+        data=post_data,
+        cookies=cookies,
+    )
+    assert response.status_code == 302
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "new_name,should_work,expected_message",