Permissions (#87)

* Lint and sort imports

* Add wagtailtransfer_can_import permission

* Update README.md

README.md

@@ -42,6 +42,7 @@ Developed by [Torchbox](https://torchbox.com/) and sponsored by [The Motley Fool
 
   `WAGTAILTRANSFER_SECRET_KEY` and the per-source `SECRET_KEY` settings are used to authenticate the communication between the source and destination instances; this prevents unauthorised users from using this API to retrieve sensitive data such as password hashes. The `SECRET_KEY` for each entry in `WAGTAILTRANSFER_SOURCES` must match that instance's `WAGTAILTRANSFER_SECRET_KEY`.
 
+* Create a user group (Wagtail admin > Settings > Groups > Add a group) with the permission "Can import pages and snippets from other sites". The "Import" menu item, and the associated import views, will only be available to members of this group (and superusers).
 
 ## Configuration
 

tests/tests/test_views.py

@@ -1,11 +1,15 @@
 import json
+from datetime import date, datetime, timezone
 from unittest import mock
-from datetime import datetime, date, timezone
 
+from django.contrib.auth.models import AnonymousUser, Group, Permission, User
+from django.contrib.contenttypes.models import ContentType
+from django.shortcuts import redirect
 from django.test import TestCase
+from django.urls import reverse
 
-from wagtail_transfer.operations import ImportPlanner
-from tests.models import PageWithRichText, SectionedPage, SimplePage, SponsoredPage
+from tests.models import SponsoredPage
+from wagtail_transfer.models import IDMapping
 
 
 class TestChooseView(TestCase):
@@ -286,3 +290,138 @@ def test_list_snippet_models(self, get, post):
         snippet = content['items'][0]
         self.assertEqual(snippet['model_label'], 'tests.category')
         self.assertEqual(snippet['name'], 'Category')
+
+
+class ImportPermissionsTests(TestCase):
+    fixtures = ["test.json"]
+
+    def setUp(self):
+        idmapping_content_type = ContentType.objects.get_for_model(IDMapping)
+        can_import_permission = Permission.objects.get(
+            content_type=idmapping_content_type, codename="wagtailtransfer_can_import",
+        )
+        can_access_admin_permission = Permission.objects.get(
+            content_type=ContentType.objects.get(
+                app_label="wagtailadmin", model="admin",
+            ),
+            codename="access_admin",
+        )
+
+        page_importers_group = Group.objects.create(name="Page importers")
+        page_importers_group.permissions.add(can_import_permission)
+        page_importers_group.permissions.add(can_access_admin_permission)
+
+        editors = Group.objects.get(name="Editors")
+
+        self.superuser = User.objects.create_superuser(
+            username="superuser", email="[email protected]", password="password",
+        )
+        self.inactive_superuser = User.objects.create_superuser(
+            username="inactivesuperuser",
+            email="[email protected]",
+            password="password",
+        )
+        self.inactive_superuser.is_active = False
+        self.inactive_superuser.save()
+
+        # a user with can_import_pages permission through the 'Page importers' group
+        self.page_importer = User.objects.create_user(
+            username="pageimporter",
+            email="[email protected]",
+            password="password",
+        )
+        self.page_importer.groups.add(page_importers_group)
+        self.page_importer.groups.add(editors)
+
+        # a user with can_import_pages permission through user_permissions
+        self.oneoff_page_importer = User.objects.create_user(
+            username="oneoffpageimporter",
+            email="[email protected]",
+            password="password",
+        )
+        self.oneoff_page_importer.user_permissions.add(can_import_permission)
+        self.oneoff_page_importer.user_permissions.add(can_access_admin_permission)
+        self.oneoff_page_importer.groups.add(editors)
+
+        # a user with can_import_pages permission through user_permissions
+        self.vanilla_user = User.objects.create_user(
+            username="vanillauser", email="[email protected]", password="password"
+        )
+        self.vanilla_user.user_permissions.add(can_access_admin_permission)
+
+        # a user that has can_import_pages permission, but is inactive
+        self.inactive_page_importer = User.objects.create_user(
+            username="inactivepageimporter",
+            email="[email protected]",
+            password="password",
+        )
+        self.inactive_page_importer.groups.add(page_importers_group)
+        self.inactive_page_importer.groups.add(editors)
+        self.inactive_page_importer.is_active = False
+        self.inactive_page_importer.save()
+
+        self.anonymous_user = AnonymousUser()
+
+        self.permitted_users = [
+            self.superuser,
+            self.page_importer,
+            self.oneoff_page_importer,
+        ]
+        self.denied_users = [
+            self.anonymous_user,
+            self.inactive_superuser,
+            self.inactive_page_importer,
+            self.vanilla_user,
+        ]
+
+    def _test_view(self, method, url, data=None, success_url=None):
+
+        for user in self.permitted_users:
+            with self.subTest(user=user):
+                self.client.login(username=user.username, password="password")
+                request = getattr(self.client, method)
+                response = request(url, data)
+                if success_url:
+                    self.assertRedirects(response, success_url)
+                else:
+                    self.assertEqual(response.status_code, 200)
+            self.client.logout()
+
+        for user in self.denied_users:
+            with self.subTest(user=user):
+                if user.is_authenticated:
+                    self.client.login(username=user.username, password="password")
+
+                request = getattr(self.client, method)
+                response = request(url, data)
+                self.assertEqual(response.status_code, 302)
+                if user == self.vanilla_user:
+                    # expect redirect loop; cf. https://github.com/wagtail/wagtail/issues/431
+                    self.assertEqual(response.status_code, 302)
+                    self.assertEqual(
+                        response.url, reverse("wagtailadmin_login") + f"?next={url}"
+                    )
+                else:
+                    self.assertRedirects(
+                        response, reverse("wagtailadmin_login") + f"?next={url}"
+                    )
+            self.client.logout()
+
+    def test_chooser_view(self):
+        url = "/admin/wagtail-transfer/choose/"
+        method = "get"
+        self._test_view(method, url)
+
+    @mock.patch("wagtail_transfer.views.import_page")
+    def test_do_import_view(self, mock_import_page):
+        success_url = "/admin/pages/2/"
+        mock_import_page.return_value = redirect(success_url)
+
+        url = "/admin/wagtail-transfer/import/"
+        method = "post"
+        data = {
+            "source": "staging",
+            "source_page_id": "12",
+            "dest_page_id": "2",
+        }
+        self._test_view(method, url, data, success_url=success_url)

wagtail_transfer/migrations/0003_permissions.py

@@ -0,0 +1,48 @@
+from django.contrib.contenttypes.management import create_contenttypes
+from django.db import migrations
+
+
+def create_import_permission(apps, schema_editor):
+    app_config = apps.get_app_config("wagtail_transfer")
+    # Ensure content types from previous migrations are created. This is normally done
+    # in a post_migrate signal, see
+    # https://github.com/django/django/blob/3.2/django/contrib/contenttypes/apps.py#L21
+    app_config.models_module = getattr(app_config, 'models_module', None) or True
+    create_contenttypes(app_config)
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    content_type = ContentType.objects.get(
+        app_label="wagtail_transfer", model="idmapping"
+    )
+    Permission = apps.get_model("auth", "Permission")
+    Permission.objects.get_or_create(
+        content_type=content_type,
+        codename="wagtailtransfer_can_import",
+        name="Can import pages and snippets from other sites",
+    )
+
+
+def delete_import_permission(apps, schema_editor):
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    content_type = ContentType.objects.get(
+        app_label="wagtail_transfer", model="idmapping"
+    )
+    Permission = apps.get_model("auth", "Permission")
+    permission = Permission.objects.filter(
+        codename="wagtailtransfer_can_import",
+        content_type=content_type,
+    )
+    permission.delete()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("auth", "0011_update_proxy_permissions"),
+        ("contenttypes", "0002_remove_content_type_name"),
+        ("wagtailcore", "0060_fix_workflow_unique_constraint"),
+        ("wagtail_transfer", "0002_importedfile"),
+    ]
+
+    operations = [
+        migrations.RunPython(create_import_permission, delete_import_permission),
+    ]

wagtail_transfer/views.py

@@ -3,6 +3,7 @@
 
 import requests
 from django.conf import settings
+from django.contrib.auth.decorators import permission_required
 from django.http import Http404, HttpResponse, JsonResponse
 from django.shortcuts import get_object_or_404, redirect, render
 from django.urls import reverse
@@ -180,6 +181,9 @@ class PageChooserAPIViewSet(PagesAdminAPIViewSet):
     ]
 
 
+@permission_required(
+    "wagtail_transfer.wagtailtransfer_can_import", login_url="wagtailadmin_login"
+)
 def chooser_api_proxy(request, source_name, path):
     source_config = getattr(settings, 'WAGTAILTRANSFER_SOURCES', {}).get(source_name)
 
@@ -201,6 +205,9 @@ def chooser_api_proxy(request, source_name, path):
     return HttpResponse(response.content, status=response.status_code)
 
 
+@permission_required(
+    "wagtail_transfer.wagtailtransfer_can_import", login_url="wagtailadmin_login"
+)
 def choose_page(request):
     return render(request, 'wagtail_transfer/choose_page.html', {
         'sources_data': json.dumps([
@@ -277,6 +284,9 @@ def import_model(request):
     return redirect('wagtailsnippets:list', app_label, model_name)
 
 
+@permission_required(
+    "wagtail_transfer.wagtailtransfer_can_import", login_url="wagtailadmin_login"
+)
 @require_POST
 def do_import(request):
     post_type = request.POST.get('type', 'page')

wagtail_transfer/wagtail_hooks.py

@@ -3,11 +3,10 @@
 from django.urls import reverse
 from wagtail.admin.menu import MenuItem
 from wagtail.core import hooks
+from django.contrib.auth.models import Permission
 
 from . import admin_urls
 
-from django.utils.html import format_html
-
 try:
     # Django 2
     from django.contrib.staticfiles.templatetags.staticfiles import static
@@ -25,9 +24,22 @@ def register_admin_urls():
 
 class WagtailTransferMenuItem(MenuItem):
     def is_shown(self, request):
-        return bool(getattr(settings, 'WAGTAILTRANSFER_SOURCES', None))
+        return all(
+            [
+                bool(getattr(settings, "WAGTAILTRANSFER_SOURCES", None)),
+                request.user.has_perm("wagtail_transfer.wagtailtransfer_can_import"),
+            ]
+        )
 
 
 @hooks.register('register_admin_menu_item')
 def register_admin_menu_item():
     return WagtailTransferMenuItem('Import', reverse('wagtail_transfer_admin:choose_page'), classnames='icon icon-doc-empty-inverse', order=10000)
+
+
+@hooks.register("register_permissions")
+def register_wagtail_transfer_permission():
+    return Permission.objects.filter(
+        content_type__app_label="wagtail_transfer",
+        codename="wagtailtransfer_can_import",
+    )