[layerindex-web,2/4] layerindex: restrict bulk changes to their owner

diff --git a/layerindex/urls.py b/layerindex/urls.py
index 51a832d..92f419f 100644
--- a/layerindex/urls.py
+++ b/layerindex/urls.py
@@ -6,18 +6,18 @@ 
 #
 # SPDX-License-Identifier: MIT
 
-from django.views.generic import TemplateView, DetailView, ListView, RedirectView
+from django.views.generic import TemplateView, ListView, RedirectView
 from django.views.defaults import page_not_found
 from django.urls import include, re_path, reverse_lazy
 from layerindex.views import LayerListView, LayerReviewListView, LayerReviewDetailView, RecipeSearchView, \
     MachineSearchView, LayerDetailView, edit_layer_view, delete_layer_view, edit_layernote_view, delete_layernote_view, \
     HistoryListView, EditProfileFormView, AdvancedRecipeSearchView, BulkChangeView, BulkChangeSearchView, \
-    bulk_change_edit_view, bulk_change_patch_view, BulkChangeDeleteView, RecipeDetailView, RedirectParamsView, \
+    bulk_change_edit_view, bulk_change_patch_view, BulkChangeReviewView, BulkChangeDeleteView, RecipeDetailView, RedirectParamsView, \
     ClassicRecipeSearchView, ClassicRecipeDetailView, ClassicRecipeStatsView, LayerUpdateDetailView, UpdateListView, \
     UpdateDetailView, StatsView, publish_view, LayerCheckListView, BBClassCheckListView, TaskStatusView, \
     ComparisonRecipeSelectView, ComparisonRecipeSelectDetailView, task_log_view, task_stop_view, email_test_view, \
     BranchCompareView, RecipeDependenciesView, update_layer_view
-from layerindex.models import LayerItem, Recipe, RecipeChangeset
+from layerindex.models import LayerItem, Recipe
 from rest_framework import routers
 from . import restviews
 
@@ -97,9 +97,7 @@  urlpatterns = [
     re_path(r'^bulkchange/(?P<pk>\d+)/edit/$',
         bulk_change_edit_view, {'template_name': 'layerindex/bulkchangeedit.html'}, name="bulk_change_edit"),
     re_path(r'^bulkchange/(?P<pk>\d+)/review/$',
-        DetailView.as_view(
-            model=RecipeChangeset,
-            context_object_name='changeset',
+        BulkChangeReviewView.as_view(
             template_name='layerindex/bulkchangereview.html'),
         name="bulk_change_review"),
     re_path(r'^bulkchange/(?P<pk>\d+)/patches/$',
diff --git a/layerindex/views.py b/layerindex/views.py
index c030ae1..42722bf 100644
--- a/layerindex/views.py
+++ b/layerindex/views.py
@@ -307,8 +307,9 @@  def edit_layer_view(request, template_name, branch='master', slug=None):
         'return_url': return_url,
     })
 
+@login_required
 def bulk_change_edit_view(request, template_name, pk):
-    changeset = get_object_or_404(RecipeChangeset, pk=pk)
+    changeset = get_object_or_404(RecipeChangeset, pk=pk, user=request.user)
 
     if request.method == 'POST':
         formset = BulkChangeEditFormSet(request.POST, queryset=changeset.recipechange_set.all())
@@ -322,8 +323,9 @@  def bulk_change_edit_view(request, template_name, pk):
         'formset': formset,
     })
 
+@login_required
 def bulk_change_patch_view(request, pk):
-    changeset = get_object_or_404(RecipeChangeset, pk=pk)
+    changeset = get_object_or_404(RecipeChangeset, pk=pk, user=request.user)
     # FIXME this couples the web server and machine running the update script together,
     # but given that it's a separate script the way is open to decouple them in future
     try:
@@ -855,10 +857,27 @@  class BaseDeleteView(DeleteView):
         return context
 
 
+class BulkChangeReviewView(DetailView):
+    model = RecipeChangeset
+    context_object_name = 'changeset'
+
+    @method_decorator(login_required)
+    def dispatch(self, request, *args, **kwargs):
+        return super(BulkChangeReviewView, self).dispatch(request, *args, **kwargs)
+
+    def get_queryset(self):
+        qs = super(BulkChangeReviewView, self).get_queryset()
+        return qs.filter(user=self.request.user)
+
+
 class BulkChangeDeleteView(BaseDeleteView):
     model = RecipeChangeset
     success_url = reverse_lazy('bulk_change')
 
+    @method_decorator(login_required)
+    def dispatch(self, request, *args, **kwargs):
+        return super(BulkChangeDeleteView, self).dispatch(request, *args, **kwargs)
+
     def get_queryset(self):
         qs = super(BulkChangeDeleteView, self).get_queryset()
         return qs.filter(user=self.request.user)
diff --git a/tests/test_bulkchange_access.py b/tests/test_bulkchange_access.py
new file mode 100644
index 0000000..e85451d
--- /dev/null
+++ b/tests/test_bulkchange_access.py
@@ -0,0 +1,53 @@ 
+# layerindex-web - tests for bulk change access control
+#
+# Licensed under the MIT license, see COPYING.MIT for details
+#
+# SPDX-License-Identifier: MIT
+
+import pytest
+from django.contrib.auth.models import User
+from django.test import TestCase
+from django.urls import reverse
+
+from layerindex.models import RecipeChangeset
+
+
+@pytest.mark.django_db
+class TestBulkChangeAccess(TestCase):
+
+    def setUp(self):
+        self.owner = User.objects.create_user('owner', 'owner@example.com', 'pw')
+        self.other = User.objects.create_user('other', 'other@example.com', 'pw')
+        self.changeset = RecipeChangeset.objects.create(user=self.owner, name='mine')
+
+    def _url(self, name):
+        return reverse(name, args=(self.changeset.pk,))
+
+    def test_anonymous_redirected_to_login(self):
+        for name in ('bulk_change_edit', 'bulk_change_review', 'bulk_change_patches', 'bulk_change_delete'):
+            response = self.client.get(self._url(name))
+            self.assertEqual(response.status_code, 302, name)
+            self.assertIn('login', response['Location'], name)
+
+    def test_other_user_gets_404(self):
+        self.client.force_login(self.other)
+        for name in ('bulk_change_edit', 'bulk_change_review', 'bulk_change_patches', 'bulk_change_delete'):
+            response = self.client.get(self._url(name))
+            self.assertEqual(response.status_code, 404, name)
+        response = self.client.post(self._url('bulk_change_delete'))
+        self.assertEqual(response.status_code, 404)
+        self.assertTrue(RecipeChangeset.objects.filter(pk=self.changeset.pk).exists())
+
+    def test_other_user_cannot_post_edits(self):
+        self.client.force_login(self.other)
+        # A valid (empty) edit formset, which would be saved and redirect
+        # to the review page if the changeset were found
+        data = {'form-TOTAL_FORMS': '0', 'form-INITIAL_FORMS': '0'}
+        response = self.client.post(self._url('bulk_change_edit'), data)
+        self.assertEqual(response.status_code, 404)
+
+    def test_owner_can_access(self):
+        self.client.force_login(self.owner)
+        for name in ('bulk_change_edit', 'bulk_change_review'):
+            response = self.client.get(self._url(name))
+            self.assertEqual(response.status_code, 200, name)