Merge pull request #228 from rhyw/user-restrictions

Configurable restrictions for user list/detail view.

kobo/admin/templates/hub/settings.py.template

@@ -160,6 +160,13 @@ XMLRPC_METHODS = {
     ),
 }
 
+# Denote whether the access to user list/detail view is restricted
+# Possible values:
+# "" (empty string) = Anonymous access (default)
+# "authenticated" = Authenticated users
+# "staff" = Staff (admin) users only
+USERS_ACL_PERMISSION = ""
+
 # override default values with custom ones from local settings
 try:
     from settings_local import *

kobo/django/views/generic.py

@@ -3,10 +3,63 @@
 import warnings
 from django.conf import settings
 from django.db.models.query import QuerySet
+from django.http import HttpResponse
+from django.template.loader import get_template
 from django.views.generic.edit import ProcessFormView, FormMixin
 from django.views.generic.list import ListView
 from django.views.generic.detail import DetailView
 
+
+class UsersAclMixin:
+    """
+    Mixin class for access control on user list/detail views.
+
+    This mixin checks the `USERS_ACL_PERMISSION` in settings to determine if
+    the user list/detail view can be accessed.
+    """
+    def get_acl_permission(self):
+        return getattr(settings, "USERS_ACL_PERMISSION", "")
+
+    def _has_access(self, request):
+        """
+        Check if the user has access based on `USERS_ACL_PERMISSION` in settings.
+
+        Returns:
+            bool: True if the user has access, False otherwise.
+        """
+        acl_permission = self.get_acl_permission()
+        if acl_permission == "authenticated" and not request.user.is_authenticated:
+            return False
+        elif acl_permission == "staff" and not request.user.is_staff:
+            return False
+        return True
+
+    def get_queryset(self):
+        queryset = super().get_queryset()
+        if not self._has_access(self.request):
+            return queryset.none()
+
+        return queryset
+
+    def dispatch(self, request, *args, **kwargs):
+        if not self._has_access(request):
+            acl_permission = self.get_acl_permission()
+            if acl_permission == "staff":
+                message = "Permission denied: only staff users can access."
+            else:
+                message = "Permission denied: you must login to access."
+            template = get_template("base.html")
+            context = {
+                "error_message": message
+            }
+            return HttpResponse(
+                template.render(context, request=request),
+                status=403
+            )
+
+        return super().dispatch(request, *args, **kwargs)
+
+
 class ExtraListView(ListView):
     paginate_by = getattr(settings, "PAGINATE_BY", None)
     extra_context = None
@@ -20,6 +73,11 @@ def get_context_data(self, **kwargs):
             context['title'] = self.title
         return context
 
+
+class UserListView(UsersAclMixin, ExtraListView):
+    pass
+
+
 class ExtraDetailView(DetailView):
     extra_context = None
     title = None
@@ -32,6 +90,7 @@ def get_context_data(self, **kwargs):
             context['title'] = self.title
         return context
 
+
 class SearchView(FormMixin, ProcessFormView, ExtraListView):
     """
     Appends GET variables to the context. Used for paginated searches.
@@ -84,6 +143,7 @@ def get_context_data(self, **kwargs):
         context['get_vars'] = get_vars
         return context
 
+
 def object_list(request, **kwargs):
     warnings.warn('object_list is deprecated, please use ExtraListView or SearchView in future.')
     if 'form' in kwargs:

kobo/hub/__init__.py

@@ -30,6 +30,16 @@
         raise ImproperlyConfigured("Invalid permissions on '%s'." % dir_path)
 
 
+if hasattr(settings, "USERS_ACL_PERMISSION"):
+    acl_permission = getattr(settings, "USERS_ACL_PERMISSION")
+    valid_options = ["", "authenticated", "staff"]
+    if acl_permission not in valid_options:
+        raise ImproperlyConfigured(
+            f"Invalid USERS_ACL_PERMISSION in settings: '{acl_permission}', must be one of "
+            f"'authenticated', 'staff' or ''(empty string)"
+        )
+
+
 if getattr(settings, "MIDDLEWARE", None) is not None:
     # Settings defines Django>=1.10 style middleware, check that
     middleware_var = "MIDDLEWARE"

kobo/hub/urls/user.py

@@ -8,13 +8,13 @@
 
 else:
     from django.conf.urls import url
-from kobo.django.views.generic import ExtraListView
+from kobo.django.views.generic import UserListView
 from kobo.hub.views import UserDetailView
 from kobo.django.compat import gettext_lazy as _
 
 
 urlpatterns = [
-    url(r"^$", ExtraListView.as_view(
+    url(r"^$", UserListView.as_view(
         queryset=get_user_model().objects.order_by("username"),
         template_name="user/list.html",
         context_object_name="usr_list",

kobo/hub/views.py

@@ -21,7 +21,7 @@
 
 from kobo.hub.models import Arch, Channel, Task
 from kobo.hub.forms import TaskSearchForm
-from kobo.django.views.generic import ExtraDetailView, SearchView
+from kobo.django.views.generic import ExtraDetailView, SearchView, UsersAclMixin
 from kobo.django.compat import gettext_lazy as _
 
 
@@ -35,7 +35,7 @@
 LOG_WATCHER_INTERVAL = getattr(settings, "LOG_WATCHER_INTERVAL", 5000)
 
 
-class UserDetailView(ExtraDetailView):
+class UserDetailView(UsersAclMixin, ExtraDetailView):
     model = get_user_model()
     title = _("User detail")
     template_name = "user/detail.html"

tests/test_views.py

@@ -192,24 +192,78 @@ def test_detail(self):
 
 class TestUserView(django.test.TransactionTestCase):
 
+    def _create_user(self, username, password="test", is_staff=False):
+        user = User.objects.create(username=username)
+        user.set_password(password)
+        user.is_staff = is_staff
+        user.save()
+
+        return user
+
     def setUp(self):
         self._fixture_teardown()
-        self.user1 = User.objects.create(username='user-1')
-        self.user2 = User.objects.create(username='user-2')
+        self.user1 = self._create_user("user1")
+        self.user2 = self._create_user("user2")
+        self.staff_user = self._create_user("user3", is_staff=True)
 
         self.client = django.test.Client()
 
+    # nothing will be impacted if `USERS_ACL_PERMISSION` is not available
+    # in settings or is an empty string
+    @override_settings(USERS_ACL_PERMISSION="")
     def test_list(self):
         response = self.client.get('/info/user/')
         self.assertEqual(response.status_code, 200)
         self.assertTrue(self.user1.username in str(response.content))
         self.assertTrue(self.user2.username in str(response.content))
 
+    @override_settings(USERS_ACL_PERMISSION="")
     def test_detail(self):
         response = self.client.get('/info/user/%d/' % self.user1.id)
         self.assertEqual(response.status_code, 200)
         self.assertTrue('#%d: %s' % (self.user1.id, self.user1.username) in str(response.content))
 
+    @override_settings(USERS_ACL_PERMISSION="authenticated")
+    def test_authenticated_access_user_list(self):
+        response = self.client.get('/info/user/')
+        self.assertEqual(response.status_code, 403)
+        # user should have access once logged in
+        self.client.login(username=self.user1.username, password="test")
+        response = self.client.get('/info/user/')
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(USERS_ACL_PERMISSION="authenticated")
+    def test_authenticated_access_user_detail(self):
+        response = self.client.get('/info/user/%d/' % self.user1.id)
+        self.assertEqual(response.status_code, 403)
+        self.client.login(username=self.user1.username, password="test")
+        response = self.client.get('/info/user/%d/' % self.user1.id)
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(USERS_ACL_PERMISSION="staff")
+    def test_staff_access_user_list(self):
+        response = self.client.get('/info/user/')
+        self.assertEqual(response.status_code, 403)
+        # no access for authenticated user
+        self.client.login(username=self.user1.username, password="test")
+        response = self.client.get('/info/user/')
+        self.assertEqual(response.status_code, 403)
+        self.client.login(username=self.staff_user.username, password="test")
+        response = self.client.get('/info/user/')
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(USERS_ACL_PERMISSION="staff")
+    def test_staff_access_user_detail(self):
+        user_detail_url = '/info/user/%d/' % self.user2.id
+        response = self.client.get(user_detail_url)
+        self.assertEqual(response.status_code, 403)
+        self.client.login(username=self.user1.username, password="test")
+        response = self.client.get(user_detail_url)
+        self.assertEqual(response.status_code, 403)
+        self.client.login(username=self.staff_user.username, password="test")
+        response = self.client.get(user_detail_url)
+        self.assertEqual(response.status_code, 200)
+
 
 class TestWorkerView(django.test.TransactionTestCase):