Add Permission Checks for Promgen Web

Create a mixin class to check permission logic:
- For Service/Farm, the object being checked is itself.
- For Project, the object being checked is itself or its parent Service.
- For Host, the object being checked is its parent Farm.
- For Exporter/URL, the object being checked is its parent Project
or the Service that is the parent of the Project.
- For Rule/Sender, the object being checked is its parent Service/Project.
- Other cases only have permission if the user being checked is a superuser.

Apply the mixin class to View classes:
- 'View' classes (List, Detail) are not applied.
- The 'ServiceRegister' class is not applied.
- 'Update' classes require the user to have EDIT or MANAGE permissions.
- 'Delete' classes require the user to have MANAGE permissions.

Author: hoangpn

promgen/mixins.py

@@ -1,11 +1,12 @@
 # Copyright (c) 2019 LINE Corporation
 # These sources are released under the terms of the MIT license: see LICENSE
-
+import guardian.mixins
+import guardian.utils
 from django.contrib import messages
 from django.contrib.auth.mixins import PermissionRequiredMixin
 from django.contrib.auth.views import redirect_to_login
 from django.contrib.contenttypes.models import ContentType
-from django.shortcuts import get_object_or_404
+from django.shortcuts import get_object_or_404, redirect
 from django.views.generic.base import ContextMixin
 
 from promgen import models
@@ -78,3 +79,67 @@ def get_context_data(self, **kwargs):
                 models.Service, id=self.kwargs["pk"]
             )
         return context
+
+
+class PromgenGuardianPermissionMixin(guardian.mixins.PermissionRequiredMixin):
+
+    def get_check_permission_object(self):
+        # Override this method to return the object to check permissions for
+        return self.get_object()
+
+    def get_check_permission_objects(self):
+        # We only define permission for Service/Project/Farm
+        # So we need to check the permission for the parent objects in other cases
+        try:
+            object = self.get_check_permission_object()
+            if isinstance(object, models.Farm):
+                return [object]
+            elif isinstance(object, models.Host):
+                return [object, object.farm]
+            elif isinstance(object, models.Service):
+                return [object]
+            elif isinstance(object, models.Project):
+                return [object, object.service]
+            elif isinstance(object, models.Exporter) or isinstance(object, models.URL):
+                return [object.project, object.project.service]
+            elif isinstance(object, models.Rule) or isinstance(object, models.Sender):
+                if isinstance(object.content_object, models.Project):
+                    return [object.content_object, object.content_object.service]
+                else:
+                    return [object.content_object]
+        except:
+            return None
+
+    def check_permissions(self, request):
+        check_permission_objects = self.get_check_permission_objects()
+        if check_permission_objects is None:
+            if request.user.is_active and request.user.is_superuser:
+                return None
+            return self.on_permission_check_fail(request, None)
+        # Loop through all the objects to check permissions for
+        # If any of the objects has the required permission (any_perm=True), we can proceed
+        # Otherwise, we will return the forbidden response
+        forbidden = None
+        for obj in check_permission_objects:
+            forbidden = guardian.utils.get_40x_or_None(request,
+                                                       perms=self.get_required_permissions(
+                                                           request),
+                                                       obj=obj,
+                                                       login_url=self.login_url,
+                                                       redirect_field_name=self.redirect_field_name,
+                                                       return_403=self.return_403,
+                                                       return_404=self.return_404,
+                                                       accept_global_perms=False,
+                                                       any_perm=True,
+                                                       )
+            if forbidden is None:
+                break
+        if forbidden:
+            return self.on_permission_check_fail(request, forbidden)
+
+    def on_permission_check_fail(self, request, response, obj=None):
+        messages.warning(request, "You do not have permission to perform this action.")
+        referer = request.META.get("HTTP_REFERER")
+        if referer:
+            return redirect(referer)
+        return redirect_to_login(self.request.get_full_path())

promgen/tests/test_host_add.py

@@ -1,10 +1,11 @@
 # Copyright (c) 2017 LINE Corporation
 # These sources are released under the terms of the MIT license: see LICENSE
-
-
+from django.shortcuts import get_object_or_404
 from django.urls import reverse
+from guardian.shortcuts import assign_perm
 
 from promgen import models, validators
+from promgen.middleware import get_current_user
 from promgen.tests import PromgenTest
 
 
@@ -16,6 +17,7 @@ def setUp(self):
     # separated and comma separated work, but are not necessarily testing
     # valid/invalid hostnames
     def test_newline(self):
+        assign_perm("promgen.edit_farm", get_current_user(), get_object_or_404(models.Farm, pk=1))
         self.client.post(
             reverse("hosts-add", args=[1]),
             {"hosts": "\naaa.example.com\nbbb.example.com\nccc.example.com \n"},
@@ -24,6 +26,7 @@ def test_newline(self):
         self.assertCount(models.Host, 3, "Expected 3 hosts")
 
     def test_comma(self):
+        assign_perm("promgen.edit_farm", get_current_user(), get_object_or_404(models.Farm, pk=1))
         self.client.post(
             reverse("hosts-add", args=[1]),
             {"hosts": ",,aaa.example.com, bbb.example.com,ccc.example.com,"},

promgen/tests/test_mixins.py

@@ -0,0 +1,76 @@
+# Copyright (c) 2025 LINE Corporation
+# These sources are released under the terms of the MIT license: see LICENSE
+from unittest.mock import patch
+
+from django.shortcuts import get_object_or_404
+from django.test import RequestFactory
+from guardian.shortcuts import assign_perm
+
+from promgen import models
+from promgen import tests
+from promgen.mixins import PromgenGuardianPermissionMixin
+
+
+class MockView(PromgenGuardianPermissionMixin):
+    def get_object(self):
+        return self.object
+
+    def dispatch(self, request, *args, **kwargs):
+        self.request = request
+        response = self.check_permissions(request)
+        if response:
+            return "Permission Denied"
+        return "Permission Granted"
+
+
+class PromgenGuardianPermissionMixinTest(tests.PromgenTest):
+    def setUp(self):
+        self.view = MockView()
+        factory = RequestFactory()
+        self.request = factory.get("/example-url")
+
+    def test_permission_granted(self):
+        user = self.force_login(username="demo")
+        object = get_object_or_404(models.Project, pk=1)
+        permission_required = "manage_project"
+        assign_perm(permission_required, user, object)
+        self.view.permission_required = permission_required
+        self.view.object = object
+        self.request.user = user
+        response = self.view.dispatch(self.request)
+        self.assertEqual(response, "Permission Granted")
+
+    @patch("django.contrib.messages.api.add_message")
+    def test_permission_not_granted(self, mock_add_message):
+        user = self.force_login(username="demo")
+        object = get_object_or_404(models.Project, pk=1)
+        permission_required = "manage_project"
+        self.view.permission_required = permission_required
+        self.view.object = object
+        self.request.user = user
+        response = self.view.dispatch(self.request)
+        self.assertEqual(response, "Permission Denied")
+
+    def test_permission_granted_on_parent_object(self):
+        user = self.force_login(username="demo")
+        object = get_object_or_404(models.Service, pk=1)
+        permission_required = "manage_service"
+        assign_perm(permission_required, user, object)
+        self.view.permission_required = permission_required
+        self.view.object = object
+        self.request.user = user
+        response = self.view.dispatch(self.request)
+        self.assertEqual(response, "Permission Granted")
+
+    @patch("django.contrib.messages.api.add_message")
+    def test_permission_granted_on_another_object(self, mock_add_message):
+        user = self.force_login(username="demo")
+        object = get_object_or_404(models.Service, pk=1)
+        another_object = models.Service.objects.create(name="Another Service")
+        permission_required = "manage_service"
+        assign_perm(permission_required, user, another_object)
+        self.view.permission_required = permission_required
+        self.view.object = object
+        self.request.user = user
+        response = self.view.dispatch(self.request)
+        self.assertEqual(response, "Permission Denied")

promgen/tests/test_routes.py

@@ -7,6 +7,7 @@
 from django.urls import reverse
 
 from promgen import models, tests, views
+from promgen.middleware import get_current_user
 
 TEST_SETTINGS = tests.Data("examples", "promgen.yml").yaml()
 TEST_IMPORT = tests.Data("examples", "import.json").raw()
@@ -104,7 +105,11 @@ def test_failed_permission(self):
             self.assertTrue(response.url.startswith("/login"))
 
     def test_other_routes(self):
-        self.add_user_permissions("promgen.add_rule", "promgen.change_site")
+        user = get_current_user()
+        user.is_superuser = True
+        user.save()
         for request in [{"viewname": "rule-new", "args": ("site", 1)}]:
             response = self.client.get(reverse(**request))
             self.assertRoute(response, views.AlertRuleRegister, 200)
+        user.is_superuser = False
+        user.save()

promgen/tests/test_web.py

@@ -1,11 +1,13 @@
 # Copyright (c) 2022 LINE Corporation
 # These sources are released under the terms of the MIT license: see LICENSE
 from django.urls import reverse
+from guardian.shortcuts import assign_perm, remove_perm
 
-from promgen import tests, views
+from promgen import views, models
+from promgen.tests import PromgenTest
 
 
-class WebTests(tests.PromgenTest):
+class WebTests(PromgenTest):
     fixtures = ["testcases.yaml", "extras.yaml"]
 
     route_map = [
@@ -15,9 +17,31 @@ class WebTests(tests.PromgenTest):
         ("service-list", views.ServiceList, {}),
         ("service-detail", views.ServiceDetail, {"pk": 1}),
         ("project-detail", views.ProjectDetail, {"pk": 1}),
-        ("farm-link", views.FarmLink, {"pk": 1, "source": "promgen"}),
-        ("project-exporter", views.ExporterRegister, {"pk": 1}),
-        ("project-notifier", views.ProjectNotifierRegister, {"pk": 1}),
+        ("farm-link", views.FarmLink,
+         {
+             "pk": 1,
+             "source": "promgen",
+             "permission": "edit_project",
+             "model": models.Project,
+             "permission_object_pk": 1
+         }
+         ),
+        ("project-exporter", views.ExporterRegister,
+         {
+             "pk": 1,
+             "permission": "edit_project",
+             "model": models.Project,
+             "permission_object_pk": 1
+         }
+         ),
+        ("project-notifier", views.ProjectNotifierRegister,
+         {
+             "pk": 1,
+             "permission": "edit_project",
+             "model": models.Project,
+             "permission_object_pk": 1
+         }
+         ),
         ("url-list", views.URLList, {}),
         ("farm-list", views.FarmList, {}),
         ("farm-detail", views.FarmDetail, {"pk": 1}),
@@ -40,6 +64,13 @@ def setUp(self):
 
     def test_routes(self):
         for viewname, viewclass, params in self.route_map:
+            permission = params.pop("permission", None)
+            permission_model = params.pop("model", None)
+            permission_object_pk = params.pop("permission_object_pk", None)
+            if permission and permission_model and permission_object_pk:
+                permission_object = permission_model.objects.get(pk=permission_object_pk)
+                assign_perm(permission, self.user, permission_object)
+
             # By default we'll pass all params as-is to our reverse()
             # method, but we may have a few special ones (like status_code)
             # that we want to pop and handle separately
@@ -49,3 +80,7 @@ def test_routes(self):
             with self.subTest(viewname=viewname, params=params):
                 response = self.client.get(reverse(viewname, kwargs=params))
                 self.assertRoute(response, viewclass, status_code)
+
+            if permission and permission_model and permission_object_pk:
+                permission_object = permission_model.objects.get(pk=permission_object_pk)
+                remove_perm(permission, self.user, permission_object)