[fix] Ensure admin actions are executed only if user has permission

django_x509/base/admin.py

@@ -1,5 +1,6 @@
 from django import forms
-from django.contrib.admin import ModelAdmin
+from django.contrib import messages
+from django.contrib.admin import ModelAdmin, action
 from django.http import HttpResponse
 from django.shortcuts import get_object_or_404, render
 from django.urls import path, reverse
@@ -148,6 +149,7 @@ def crl_view(self, request, pk):
             instance.crl, status=200, content_type='application/x-pem-file'
         )
 
+    @action(description=_('Renew selected CAs'), permissions=['change'])
     def renew_ca(self, request, queryset):
         if request.POST.get('post'):
             renewed_rows = 0
@@ -165,7 +167,7 @@ def renew_ca(self, request, queryset):
                 ),
                 renewed_rows,
             ) % {'renewed_rows': renewed_rows}
-            self.message_user(request, message)
+            self.message_user(request, message, level=messages.SUCCESS)
         else:
             data = dict()
             ca_count = 0
@@ -183,8 +185,6 @@ def renew_ca(self, request, queryset):
                 ),
             )
 
-    renew_ca.short_description = _('Renew selected CAs')
-
 
 class AbstractCertAdmin(BaseAdmin):
     list_filter = ['ca', 'revoked', 'key_length', 'digest', 'created']
@@ -230,6 +230,7 @@ def ca_url(self, obj):
 
     ca_url.short_description = 'CA'
 
+    @action(description=_('Revoke selected certificates'), permissions=['change'])
     def revoke_action(self, request, queryset):
         rows = 0
         for cert in queryset:
@@ -240,10 +241,9 @@ def revoke_action(self, request, queryset):
         else:
             bit = '{0} certificates were'.format(rows)
         message = '{0} revoked.'.format(bit)
-        self.message_user(request, _(message))
-
-    revoke_action.short_description = _('Revoke selected certificates')
+        self.message_user(request, _(message), level=messages.SUCCESS)
 
+    @action(description=_('Renew selected certificates'), permissions=['change'])
     def renew_cert(self, request, queryset):
         if request.POST.get('post'):
             renewed_rows = 0
@@ -255,16 +255,14 @@ def renew_cert(self, request, queryset):
                 '%(renewed_rows)d Certificates have been successfully renewed',
                 renewed_rows,
             ) % {'renewed_rows': renewed_rows}
-            self.message_user(request, message)
+            self.message_user(request, message, level=messages.SUCCESS)
         else:
             return render(
                 request,
                 'admin/django_x509/renew_confirmation.html',
                 context=self.get_context(queryset, cert_count=len(queryset)),
             )
 
-    renew_cert.short_description = _('Renew selected certificates')
-
 
 # For backward compatibility
 CaAdmin = AbstractCaAdmin

django_x509/tests/test_admin.py

@@ -2,11 +2,15 @@
 
 from django.contrib import admin
 from django.contrib.admin.sites import AdminSite
+from django.contrib.auth import get_user_model
 from django.test import TestCase
+from django.urls import reverse
+from openwisp_utils.tests import AdminActionPermTestMixin
 from swapper import load_model
 
 from . import MessagingRequest
 
+User = get_user_model()
 Ca = load_model('django_x509', 'Ca')
 Cert = load_model('django_x509', 'Cert')
 
@@ -101,7 +105,7 @@ def has_perm(self, perm):
 ]
 
 
-class ModelAdminTests(TestCase):
+class ModelAdminTests(AdminActionPermTestMixin, TestCase):
     app_label = 'django_x509'
     ca_admin = admin.site._registry[Ca].__class__
     cert_admin = admin.site._registry[Cert].__class__
@@ -187,6 +191,22 @@ def test_revoke_action(self):
         self.assertEqual(len(m), 1)
         self.assertEqual(str(m[0]), '1 certificate was revoked.')
 
+    def test_revoke_action_perms(self):
+        user = User.objects.create(
+            username='tester', email='[email protected]', is_staff=True
+        )
+        cert = Cert.objects.create(
+            name='test_cert', ca=Ca.objects.create(name='test_ca')
+        )
+        self._test_action_permission(
+            path=reverse(f'admin:{self.app_label}_cert_changelist'),
+            action='revoke_action',
+            user=user,
+            obj=cert,
+            message='1 certificate was revoked.',
+            required_perms=['change'],
+        )
+
     def test_renew_ca_action(self):
         req = deepcopy(request)
         ca = Ca.objects.create(name='test_ca')
@@ -221,6 +241,24 @@ def test_renew_ca_action(self):
             '1 CA and its related certificates have been successfully renewed',
         )
 
+    def test_renew_ca_action_perms(self):
+        """
+        This test verifies that only users with change permission
+        can perform the renew operation for CAs and certificates.
+        """
+        user = User.objects.create(
+            username='tester', email='[email protected]', is_staff=True
+        )
+        ca = Ca.objects.create(name='test_ca')
+        self._test_action_permission(
+            path=reverse(f'admin:{self.app_label}_ca_changelist'),
+            action='renew_ca',
+            user=user,
+            obj=ca,
+            message='1 CA and its related certificates have been successfully renewed',
+            required_perms=['change'],
+        )
+
     def test_renew_cert_action(self):
         req = deepcopy(request)
         ca = Ca.objects.create(name='test_ca')
@@ -250,3 +288,19 @@ def test_renew_cert_action(self):
         self.assertEqual(old_ca_serial_number, ca.serial_number)
         self.assertEqual(len(message), 1)
         self.assertEqual(message[0], '1 Certificate has been successfully renewed')
+
+    def test_renew_cert_action_perms(self):
+        user = User.objects.create(
+            username='tester', email='[email protected]', is_staff=True
+        )
+        cert = Cert.objects.create(
+            name='test_cert', ca=Ca.objects.create(name='test_ca')
+        )
+        self._test_action_permission(
+            path=reverse(f'admin:{self.app_label}_cert_changelist'),
+            action='renew_cert',
+            user=user,
+            obj=cert,
+            message='1 Certificate has been successfully renewed',
+            required_perms=['change'],
+        )