[feature] Allowed ready only access of shared objects to non-superusers #238

Closes #238

Author: pandafy

docs/user/basic-concepts.rst

@@ -157,6 +157,8 @@ different roles in each.
 
 Here's a summary of the default organization roles.
 
+.. _users_organization_manager:
+
 Organization Manager
 ~~~~~~~~~~~~~~~~~~~~
 
@@ -241,6 +243,18 @@ objects are defined and managed by super administrators and can include
 configurations, policies, or other data that need to be consistent across
 all organizations.
 
+Shared objects can only be created by superusers. Non-superusers (e.g.
+:ref:`users_organization_manager`) have view-only access to shared
+objects, both through the admin interface and the REST API. However, they
+can use these shared objects when creating related organization-specific
+resources. For example, an organization manager can use a shared VPN
+server to create a configuration template for their organization.
+
+In some cases, non-superusers may be restricted from viewing sensitive
+details of shared objects—particularly if such information could allow
+them to gain unauthorized access to infrastructure or data used by other
+organizations.
+
 By sharing common resources, global uniformity and consistency can be
 enforced across the entire system.
 

openwisp_users/multitenancy.py

@@ -49,14 +49,49 @@ def get_queryset(self, request):
         if user.is_superuser:
             return qs
         if hasattr(self.model, "organization"):
-            return qs.filter(organization__in=user.organizations_managed)
+            return qs.filter(
+                Q(organization__in=user.organizations_managed) | Q(organization=None)
+            )
         if self.model.__name__ == "Organization":
             return qs.filter(pk__in=user.organizations_managed)
         elif not self.multitenant_parent:
             return qs
         else:
-            qsarg = "{0}__organization__in".format(self.multitenant_parent)
-            return qs.filter(**{qsarg: user.organizations_managed})
+            qsarg = f"{self.multitenant_parent}__organization"
+            return qs.filter(
+                Q(**{f"{qsarg}__in": user.organizations_managed}) | Q(**{qsarg: None})
+            )
+
+    def _has_org_permission(self, request, obj, perm_func):
+        """
+        Helper method to check object-level permissions for users
+        associated with specific organizations.
+        """
+        perm = perm_func(request, obj)
+        if obj and self.multitenant_parent:
+            # In case of a multitenant parent, we need to check if the
+            # user has permission on the parent object.
+            obj = getattr(obj, self.multitenant_parent)
+        if not request.user.is_superuser and obj and hasattr(obj, "organization_id"):
+            perm = perm and (
+                obj.organization_id
+                and str(obj.organization_id) in request.user.organizations_managed
+            )
+        return perm
+
+    def has_change_permission(self, request, obj=None):
+        """
+        Returns True if the user has permission to change the object.
+        Non-superusers cannot change shared objects.
+        """
+        return self._has_org_permission(request, obj, super().has_change_permission)
+
+    def has_delete_permission(self, request, obj=None):
+        """
+        Returns True if the user has permission to delete the object.
+        Non-superusers cannot change shared objects.
+        """
+        return self._has_org_permission(request, obj, super().has_delete_permission)
 
     def _edit_form(self, request, form):
         """
@@ -149,7 +184,9 @@ class MultitenantOrgFilter(AutocompleteFilter):
     org_lookup = "id__in"
     title = _("organization")
     widget_attrs = AutocompleteFilter.widget_attrs.copy()
-    widget_attrs.update({"data-empty-label": SHARED_SYSTEMWIDE_LABEL})
+    widget_attrs.update(
+        {"data-empty-label": SHARED_SYSTEMWIDE_LABEL, "data-is-filter": "true"}
+    )
 
 
 class MultitenantRelatedOrgFilter(MultitenantOrgFilter):

openwisp_users/static/admin/js/autocomplete.js

@@ -0,0 +1,39 @@
+// Custom override of Django's autocomplete.js to add the "is_filter" parameter
+// to AJAX requests. This parameter enables the backend autocomplete view to
+// determine if the request is for filtering, allowing it to include the shared
+// option when appropriate (e.g., for filtering shared objects in the admin).
+
+"use strict";
+{
+  const $ = django.jQuery;
+
+  $.fn.djangoAdminSelect2 = function () {
+    $.each(this, function (i, element) {
+      $(element).select2({
+        ajax: {
+          data: (params) => {
+            return {
+              term: params.term,
+              page: params.page,
+              app_label: element.dataset.appLabel,
+              model_name: element.dataset.modelName,
+              field_name: element.dataset.fieldName,
+              is_filter: element.dataset.isFilter,
+            };
+          },
+        },
+      });
+    });
+    return this;
+  };
+
+  $(function () {
+    // Initialize all autocomplete widgets except the one in the template
+    // form used when a new formset is added.
+    $(".admin-autocomplete").not("[name*=__prefix__]").djangoAdminSelect2();
+  });
+
+  document.addEventListener("formset:added", (event) => {
+    $(event.target).find(".admin-autocomplete").djangoAdminSelect2();
+  });
+}

openwisp_users/tests/test_api/__init__.py

@@ -12,5 +12,159 @@ def _obtain_auth_token(self, username="operator", password="tester"):
         return response.data["token"]
 
 
-class APITestCase(TestMultitenantAdminMixin, AuthenticationMixin, TestCase):
+class TestMultitenantApiMixin(TestMultitenantAdminMixin):
+    def _test_access_shared_object(
+        self,
+        token,
+        listview_name=None,
+        listview_path=None,
+        detailview_name=None,
+        detailview_path=None,
+        create_payload=None,
+        update_payload=None,
+        expected_count=0,
+        expected_status_codes=None,
+    ):
+        auth = dict(HTTP_AUTHORIZATION=f"Bearer {token}")
+        create_payload = create_payload or {}
+        update_payload = update_payload or {}
+        expected_status_codes = expected_status_codes or {}
+
+        if listview_name or listview_path:
+            listview_path = listview_path or reverse(listview_name)
+            with self.subTest("HEAD and OPTION methods"):
+                response = self.client.head(listview_path, **auth)
+                self.assertEqual(response.status_code, expected_status_codes["head"])
+
+                response = self.client.options(listview_path, **auth)
+                self.assertEqual(response.status_code, expected_status_codes["option"])
+
+            with self.subTest("Create shared object"):
+                response = self.client.post(
+                    listview_path,
+                    data=create_payload,
+                    content_type="application/json",
+                    **auth,
+                )
+                self.assertEqual(response.status_code, expected_status_codes["create"])
+                if expected_status_codes["create"] == 400:
+                    self.assertEqual(
+                        str(response.data["organization"][0]),
+                        "This field may not be null.",
+                    )
+
+            with self.subTest("List all shared objects"):
+                response = self.client.get(listview_path, **auth)
+                self.assertEqual(response.status_code, expected_status_codes["list"])
+                data = response.data
+                if not isinstance(response.data, list):
+                    data = data.get("results", data)
+                self.assertEqual(len(data), expected_count)
+
+        if detailview_name or detailview_path:
+            if not detailview_path and listview_name and expected_count > 0:
+                detailview_path = reverse(detailview_name, args=[data[0]["id"]])
+
+            with self.subTest("Retrieve shared object"):
+                response = self.client.get(detailview_path, **auth)
+                self.assertEqual(
+                    response.status_code, expected_status_codes["retrieve"]
+                )
+
+            with self.subTest("Update shared object"):
+                response = self.client.put(
+                    detailview_path,
+                    data=update_payload,
+                    content_type="application/json",
+                    **auth,
+                )
+                self.assertEqual(response.status_code, expected_status_codes["update"])
+
+            with self.subTest("Delete shared object"):
+                response = self.client.delete(detailview_path, **auth)
+                self.assertEqual(response.status_code, expected_status_codes["delete"])
+
+    def _test_org_user_access_shared_object(
+        self,
+        listview_name=None,
+        listview_path=None,
+        detailview_name=None,
+        detailview_path=None,
+        create_payload=None,
+        update_payload=None,
+        expected_count=0,
+        expected_status_codes=None,
+        token=None,
+    ):
+        """
+        Non-superusers can only view shared objects.
+        They cannot create, update, or delete them.
+        """
+        if not token:
+            user = self._create_administrator(organizations=[self._get_org()])
+            token = self._obtain_auth_token(user.username, "tester")
+        if not expected_status_codes:
+            expected_status_codes = {
+                "create": 400,
+                "list": 200,
+                "retrieve": 200,
+                "update": 403,
+                "delete": 403,
+                "head": 200,
+                "option": 200,
+            }
+        self._test_access_shared_object(
+            token=token,
+            listview_name=listview_name,
+            listview_path=listview_path,
+            detailview_name=detailview_name,
+            detailview_path=detailview_path,
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=expected_count,
+            expected_status_codes=expected_status_codes,
+        )
+
+    def _test_superuser_access_shared_object(
+        self,
+        token,
+        listview_path=None,
+        listview_name=None,
+        detailview_path=None,
+        detailview_name=None,
+        create_payload=None,
+        update_payload=None,
+        expected_count=1,
+        expected_status_codes=None,
+    ):
+        """
+        Superusers can perform all operations on shared objects.
+        """
+        if not token:
+            user = self._create_admin()
+            token = self._obtain_auth_token(user.username, "tester")
+        if not expected_status_codes:
+            expected_status_codes = {
+                "create": 201,
+                "list": 200,
+                "retrieve": 200,
+                "update": 200,
+                "delete": 204,
+                "head": 200,
+                "option": 200,
+            }
+        self._test_access_shared_object(
+            token=token,
+            listview_name=listview_name,
+            listview_path=listview_path,
+            detailview_name=detailview_name,
+            detailview_path=detailview_path,
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=expected_count,
+            expected_status_codes=expected_status_codes,
+        )
+
+
+class APITestCase(TestMultitenantApiMixin, AuthenticationMixin, TestCase):
     pass

openwisp_users/tests/utils.py

@@ -1,10 +1,13 @@
 from datetime import date
 
+import django
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Permission
 from django.urls import reverse
 from swapper import load_model
 
+from openwisp_users.multitenancy import SHARED_SYSTEMWIDE_LABEL
+
 Organization = load_model("openwisp_users", "Organization")
 OrganizationOwner = load_model("openwisp_users", "OrganizationOwner")
 OrganizationUser = load_model("openwisp_users", "OrganizationUser")
@@ -236,9 +239,90 @@ def _test_recoverlist_operator_403(self, app_label, model_label):
         )
         self.assertEqual(response.status_code, 403)
 
-    def _get_autocomplete_view_path(self, app_label, model_name, field_name):
+    def _test_org_admin_create_shareable_object(
+        self,
+        path,
+        payload,
+        model,
+        expected_count=0,
+        user=None,
+        error_message=None,
+        raises_error=True,
+    ):
+        """
+        Verifies a non-superuser cannot create a shareable object
+        """
+        if not user:
+            user = self._create_administrator(organizations=[self._get_org()])
+        self.client.force_login(user)
+        response = self.client.post(
+            path,
+            data=payload,
+            follow=True,
+        )
+        if raises_error:
+            error_message = error_message or (
+                '<div class="form-row errors field-organization">\n'
+                '            <ul class="errorlist"{}>'
+                "<li>This field is required.</li></ul>"
+            ).format(' id="id_organization_error"' if django.VERSION >= (5, 2) else "")
+            self.assertContains(response, error_message)
+        self.assertEqual(model.objects.count(), expected_count)
+
+    def _test_org_admin_view_shareable_object(
+        self, path, user=None, expected_element=None
+    ):
+        """
+        Verifies a non-superuser can view a shareable object
+        """
+        if not user:
+            user = self._create_administrator(organizations=[self._get_org()])
+        self.client.force_login(user)
+        response = self.client.get(path, follow=True)
+        self.assertEqual(response.status_code, 200)
+        if not expected_element:
+            expected_element = (
+                '<div class="form-row field-organization">\n\n\n<div>\n\n'
+                '<div class="flex-container">\n\n'
+                "<label>Organization:</label>\n\n"
+                '<div class="readonly">-</div>\n\n\n'
+                "</div>\n\n</div>\n\n\n</div>"
+            )
+        self.assertContains(response, expected_element, html=True)
+
+    def _test_object_organization_fk_autocomplete_view(
+        self,
+        model,
+    ):
+        app_label = model._meta.app_label
+        model_name = model._meta.model_name
+        path = self._get_autocomplete_view_path(app_label, model_name, "organization")
+        org = self._get_org()
+        admin = User.objects.filter(is_superuser=True).first()
+        if not admin:
+            admin = self._create_admin()
+        org_admin = self._create_administrator(organizations=[org])
+
+        with self.subTest("Org admin should only see their own org"):
+            self.client.force_login(org_admin)
+            response = self.client.get(path)
+            self.assertEqual(response.status_code, 200)
+            self.assertContains(response, org.name)
+            self.assertNotContains(response, SHARED_SYSTEMWIDE_LABEL)
+
+        with self.subTest("Superuser should see all orgs and shared label"):
+            self.client.force_login(admin)
+            response = self.client.get(path)
+            self.assertEqual(response.status_code, 200)
+            self.assertContains(response, org.name)
+            self.assertContains(response, SHARED_SYSTEMWIDE_LABEL)
+
+    def _get_autocomplete_view_path(
+        self, app_label, model_name, field_name, is_filter=False
+    ):
         path = reverse("admin:ow-auto-filter")
         return (
             f"{path}?app_label={app_label}"
             f"&model_name={model_name}&field_name={field_name}"
+            "{}".format("&is_filter=true" if is_filter else "")
         )