diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 049a8bf99..e56316b7d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -75,6 +75,9 @@ jobs: pip install -U -r requirements-test.txt sudo npm install -g prettier pip install -e .[rest] + pip install --upgrade --force-reinstall --no-deps --no-cache-dir https://github.com/openwisp/openwisp-users/tarball/issues/522-disabled-org + pip install --upgrade --force-reinstall --no-deps --no-cache-dir https://github.com/openwisp/openwisp-controller/tarball/issues/1393-disabled-org + pip install --upgrade --force-reinstall --no-deps --no-cache-dir https://github.com/openwisp/openwisp-monitoring/tarball/issues/811-disabled-org pip install -U ${{ matrix.django-version }} - name: QA checks diff --git a/AGENTS.md b/AGENTS.md index eec2ab709..37bc6c784 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,6 +74,7 @@ If instructions conflict, repository config and CI workflows win first, official - Cached lookups must check permission and organization scope on every request. Changed endpoints need cross-organization regression tests. - If you change swapped-model behavior, tenant isolation, auth flows, or admin/API permissions, cover both package-level and integration tests. - Changes to HTTP REST API endpoints or Django REST Framework serializers must include tests for permissions, input validation, filtering or pagination when supported, and organization or tenant boundaries where applicable. +- Objects belonging to a disabled organization must be readable and deletable; creation and updates must be blocked across all relevant write paths. This applies to objects with either a direct or chained/nested relationship to the organization. No other operations should be permitted, except for ordinary cleanup operations. ## Troubleshooting diff --git a/docs/developer/admin-utils.rst b/docs/developer/admin-utils.rst index ea7516c0d..03e61a27a 100644 --- a/docs/developer/admin-utils.rst +++ b/docs/developer/admin-utils.rst @@ -31,6 +31,54 @@ This class has two important attributes: `_ for a real-world example. +Disabled Organization Write Protection +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +``MultitenantAdminMixin`` also blocks changes to any object belonging to a +:ref:`disabled organization `, while still +allowing it to be viewed and deleted. This also applies to superusers. For +models whose organization is reached through a parent, the mixin follows +``multitenant_parent`` to protect those objects too. + +This is controlled by the ``disabled_organization_write_protection`` class +attribute, which defaults to ``True``. Set it to ``False`` on a specific +``ModelAdmin`` to opt out: + +.. code-block:: python + + from django.contrib import admin + from openwisp_users.multitenancy import MultitenantAdminMixin + + + class BookAdmin(MultitenantAdminMixin, admin.ModelAdmin): + disabled_organization_write_protection = False + # other attributes + +The ``organization`` form field excludes disabled organizations for +everyone, including superusers. An opted-out admin keeps the object's +current disabled organization selectable so it can be saved. + +The same protection applies to **inlines** attached to a disabled object. +The parent admin denies adding and changing inline rows but keeps deletion +available, even when an inline does not use the mixin. Set +``disabled_organization_write_protection = False`` on the parent admin or +on an individual inline to opt out. + +Custom admin actions are also blocked for disabled-organization objects, +except for deletion actions. To allow a specific lifecycle action while +keeping the rest of the protection enabled, list its name in +``disabled_organization_action_exclusions``: + +.. code-block:: python + + class DeviceAdmin(MultitenantAdminMixin, admin.ModelAdmin): + disabled_organization_action_exclusions = ("deactivate_device",) + +The mixin resolves an object's organization from ``organization`` by +default and follows ``multitenant_parent`` when configured. Admins using +another relation can override ``get_object_organization()`` to return the +related organization. + ``MultitenantOrgFilter`` ------------------------ diff --git a/docs/developer/django-rest-framework-utils.rst b/docs/developer/django-rest-framework-utils.rst index 7ccc729e3..46f030401 100644 --- a/docs/developer/django-rest-framework-utils.rst +++ b/docs/developer/django-rest-framework-utils.rst @@ -131,13 +131,72 @@ organization managers or owners to view shared objects in read-only mode. Standard users will not be able to view or list shared objects. +``DisabledOrgReadOnly`` +~~~~~~~~~~~~~~~~~~~~~~~ + +**Full python path**: +``openwisp_users.api.permissions.DisabledOrgReadOnly``. + +This object-level permission class blocks updates to objects belonging to +a :ref:`disabled organization `. ``GET``, +``HEAD``, ``OPTIONS`` and ``DELETE`` remain allowed. + +The object's organization is resolved through the view's +``organization_field`` attribute, which defaults to ``"organization"``. An +invalid relation path denies the write instead of failing open. Views that +are not organization-scoped must opt out explicitly. + +.. important:: + + ``DisabledOrgReadOnly`` guards **updates only**. It implements + ``has_object_permission``, which DRF does not call on ``POST``, so it + does **not** block creation for a disabled organization. The + serializer must exclude disabled organizations from its + ``organization`` field, for example by using + ``FilterSerializerByOrgMembership``, ``FilterSerializerByOrgManaged`` + or ``FilterSerializerByOrgOwned`` mixin, or ``Organization.active``. A + plain ``ModelSerializer`` can still create an object for a disabled + organization. + +A view can opt out of this guard by setting +``allow_disabled_organization_writes = True``: + +.. code-block:: python + + from openwisp_users.api.permissions import DisabledOrgReadOnly + from rest_framework.generics import RetrieveUpdateDestroyAPIView + + + class SubnetView(RetrieveUpdateDestroyAPIView): + permission_classes = (DisabledOrgReadOnly,) + allow_disabled_organization_writes = True + +``DisabledOrgReadOnly`` is already included in ``ProtectedAPIMixin``'s +default ``permission_classes`` (see below), so views that use +``ProtectedAPIMixin`` get this guard automatically without any extra +configuration. + +.. note:: + + ``Organization.active`` (django-organizations' ``ActiveOrgManager``) + is the canonical queryset for active organizations: use + ``Organization.active.all()`` when writing custom code that needs to + select from or filter active organizations, instead of filtering + ``Organization.objects`` manually. + ``ProtectedAPIMixin`` --------------------- **Full python path**: ``openwisp_users.api.mixins.ProtectedAPIMixin``. This mixin provides a set of authentication and permission classes that -are commonly used across various OpenWISP modules API views. +are commonly used across various OpenWISP modules API views, including +``DisabledOrgReadOnly`` (see above). + +If a view overrides ``permission_classes`` entirely instead of extending +``ProtectedAPIMixin.permission_classes``, it will not inherit +``DisabledOrgReadOnly`` (or any future addition to the mixin's defaults) +automatically, and must re-declare it explicitly if the guard is needed. Usage example: @@ -255,6 +314,13 @@ and ``FilterSerializerByOrgOwned`` can be used to solve this issue. These serializers do not allow non-superusers to create shared objects. +.. _multi_tenant_serializers_disabled_org: + +These serializers also exclude :ref:`disabled organizations +` from the ``organization`` field for all +users, including superusers. Submitting a disabled organization's primary +key returns a validation error. + Usage example: .. code-block:: python diff --git a/docs/developer/misc-utils.rst b/docs/developer/misc-utils.rst index 481055f7e..bb689e4d0 100644 --- a/docs/developer/misc-utils.rst +++ b/docs/developer/misc-utils.rst @@ -311,3 +311,36 @@ Add the validator to the ``AUTH_PASSWORD_VALIDATORS`` Django setting: "NAME": "openwisp_users.password_validation.PasswordReuseValidator", }, ] + +Signals +------- + +.. include:: /partials/signals-note.rst + +``organization_disabled`` +~~~~~~~~~~~~~~~~~~~~~~~~~ + +**Path**: ``openwisp_users.signals.organization_disabled`` + +**Arguments**: + +- ``instance``: the organization instance that was disabled + +Emitted after an organization's ``is_active`` field changes from ``True`` +to ``False`` and the change has been committed to the database. + +This signal is not emitted when an organization is created. + +``organization_enabled`` +~~~~~~~~~~~~~~~~~~~~~~~~ + +**Path**: ``openwisp_users.signals.organization_enabled`` + +**Arguments**: + +- ``instance``: the organization instance that was enabled + +Emitted after an organization's ``is_active`` field changes from ``False`` +to ``True`` and the change has been committed to the database. + +This signal is not emitted when an organization is created. diff --git a/docs/user/basic-concepts.rst b/docs/user/basic-concepts.rst index f41d28126..995938e38 100644 --- a/docs/user/basic-concepts.rst +++ b/docs/user/basic-concepts.rst @@ -149,6 +149,48 @@ instance of the platform. `django-organizations `_ third-party app. +.. _users_disabled_organization: + +Disabled Organization +--------------------- + +An organization is disabled when its **Is active** flag is unchecked on +the "Change organization" page or through the REST API. Superusers and +organization managers can disable an organization, subject to the usual +permission requirements for editing it. + +A disabled organization retains its users, memberships, and related +objects. Superusers can still read and delete that data, but: + +- **No new object can be created for a disabled organization**, and + **existing objects belonging to it cannot be modified**. For the + organization itself, only **Is active** can be changed and its owner can + be unassigned while it is disabled. +- Deleting objects, including the organization itself, remains allowed. +- The organization is hidden from **organization selection widgets** but + remains available in admin **list filters**. +- Re-enabling a disabled organization is allowed **only for superusers**. + Once an organization is disabled, its managers lose access to it (a + disabled organization is no longer part of the organizations they + manage), so they can no longer edit it, including re-enabling it. A + superuser must re-enable the organization before its managers regain + access. + +.. note:: + + In the REST API, updating an object in a disabled organization returns + HTTP 400 or 403, depending on the endpoint, with an error message. + +.. note:: + + Re-enabling an organization and editing its other fields must be done + in **two separate steps**, matching the admin interface (which locks + every field except **Is active** while the organization is disabled). + First re-enable the organization (change only **Is active**), then + edit its other fields or assign an owner. A single request that both + re-enables the organization and changes another field (or assigns an + owner) is rejected. + Organization Membership and Roles --------------------------------- diff --git a/docs/user/rest-api.rst b/docs/user/rest-api.rst index 3f9a78203..47d504958 100644 --- a/docs/user/rest-api.rst +++ b/docs/user/rest-api.rst @@ -356,7 +356,9 @@ Change User Detail When editing organization memberships, the organization manager flag is represented internally by the ``is_admin`` field in the - ``organization_users`` payload. + ``organization_users`` payload. For an existing membership, omitting + ``is_admin`` leaves it unchanged, changing its value updates the role, + and sending its current value removes the membership. Patch User Detail ~~~~~~~~~~~~~~~~~ @@ -369,7 +371,9 @@ Patch User Detail When patching organization memberships, the organization manager flag is represented internally by the ``is_admin`` field in the - ``organization_users`` payload. + ``organization_users`` payload. For an existing membership, omitting + ``is_admin`` leaves it unchanged, changing its value updates the role, + and sending its current value removes the membership. Delete User ~~~~~~~~~~~ diff --git a/openwisp_users/admin.py b/openwisp_users/admin.py index 453589a68..fef01eb46 100644 --- a/openwisp_users/admin.py +++ b/openwisp_users/admin.py @@ -16,6 +16,7 @@ from django.contrib.auth.forms import UserCreationForm as BaseUserCreationForm from django.core.exceptions import ValidationError from django.db.models import Q +from django.forms.formsets import DELETION_FIELD_NAME from django.forms.models import BaseInlineFormSet from django.http import HttpResponseRedirect from django.template.response import TemplateResponse @@ -38,6 +39,7 @@ from . import settings as app_settings from .multitenancy import MultitenantAdminMixin, MultitenantOrgFilter from .utils import BaseAdmin +from .widgets import OrganizationAutocompleteSelect Group = load_model("openwisp_users", "Group") Organization = load_model("openwisp_users", "Organization") @@ -98,34 +100,115 @@ class OrganizationOwnerInline(admin.StackedInline): extra = 0 autocomplete_fields = ("organization_user",) + def has_add_permission(self, request, obj=None): + # obj is the parent Organization here + if obj is not None and not obj.is_active: + return False + return super().has_add_permission(request, obj) + def has_change_permission(self, request, obj=None): + if obj is not None and not obj.is_active: + return False if obj and not request.user.is_superuser and not request.user.is_owner(obj): return False return super().has_change_permission(request, obj) +class MultitenantReadOnlyInlineFormSet(BaseInlineFormSet): + """ + Keep rows belonging to a disabled organization valid on no-op saves while + making their editable fields read-only and preserving deletion. + """ + + organization_fk_field = "organization" + organization_lookup = "organization" + + def get_organization(self, instance): + organization = instance + for relation in self.organization_lookup.split("__"): + organization = getattr(organization, relation) + return organization + + def add_fields(self, form, index): + super().add_fields(form, index) + instance = getattr(form, "instance", None) + organization_id = getattr(instance, f"{self.organization_fk_field}_id", None) + if not (instance and instance.pk and organization_id): + return + if self.get_organization(instance).is_active: + return + organization_field = form.fields.get(self.organization_fk_field) + if organization_field is not None: + # The formset queryset excludes disabled organizations, + # so the current row's value must be added back or the + # disabled field fails validation against it. + organization_model = organization_field.queryset.model + organization_field.queryset = organization_field.queryset | ( + organization_model.objects.filter(pk=organization_id) + ) + pk_name = instance._meta.pk.name + for name, field in form.fields.items(): + # The pk field and the parent-link field must stay enabled: a + # disabled field is never submitted by the browser, and when the + # pk field is missing from POST data BaseModelFormSet treats the + # row as tampered with and silently builds a blank instance + # instead of loading the existing one, which then fails + # validation on unrelated required fields. + if name in (pk_name, self.fk.name, DELETION_FIELD_NAME): + continue + field.disabled = True + + +class OrganizationUserInlineFormSet( + MultitenantReadOnlyInlineFormSet, RequiredInlineFormSet +): + pass + + class OrganizationUserInline(admin.StackedInline): model = OrganizationUser - formset = RequiredInlineFormSet + formset = OrganizationUserInlineFormSet view_on_site = False fields = ("organization", "is_admin") autocomplete_fields = ("organization",) + def get_queryset(self, request): + # Avoid a query per inline row when checking the organization status. + return super().get_queryset(request).select_related("organization") + def get_formset(self, request, obj=None, **kwargs): """ - In form dropdowns, display only organizations - in which operator `is_admin` and for superusers - display all organizations + Limit membership choices to active organizations the user can manage. """ formset = super().get_formset(request, obj=obj, **kwargs) + org_field = formset.form.base_fields["organization"] + org_field.queryset = org_field.queryset.filter(is_active=True) if request.user.is_superuser: return formset - if not request.user.is_superuser: - formset.form.base_fields["organization"].queryset = ( - Organization.objects.filter(pk__in=request.user.organizations_managed) - ) + org_field.queryset = org_field.queryset.filter( + pk__in=request.user.organizations_managed + ) return formset + def formfield_for_foreignkey(self, db_field, request, **kwargs): + """ + Use the filtered endpoint because the stock autocomplete includes + disabled organizations. + """ + if db_field.name == "organization" and db_field.name in ( + self.get_autocomplete_fields(request) + ): + kwargs["widget"] = OrganizationAutocompleteSelect( + db_field, self.admin_site, using=kwargs.get("using") + ) + return super().formfield_for_foreignkey(db_field, request, **kwargs) + + def has_add_permission(self, request, obj=None): + # Without an active managed organization, the add form cannot be used. + if not request.user.is_superuser and not request.user.organizations_managed: + return False + return super().has_add_permission(request, obj) + def get_extra(self, request, obj=None, **kwargs): if not obj: return 1 @@ -566,7 +649,7 @@ class OrganizationAdmin( def get_inline_instances(self, request, obj=None): """ - Remove OrganizationOwnerInline from organization add form + Owners require an existing organization, so omit this inline on add. """ inlines = super().get_inline_instances(request, obj).copy() if not obj: @@ -578,12 +661,39 @@ def get_inline_instances(self, request, obj=None): def has_change_permission(self, request, obj=None): """ - Allow only managers and superuser to change organization + Keep disabled organizations accessible so superusers can re-enable them; + read-only fields enforce the remaining restrictions. """ if obj and not request.user.is_superuser and not request.user.is_manager(obj): return False + if obj and not obj.is_active: + return request.user.is_superuser return super().has_change_permission(request, obj) + def get_readonly_fields(self, request, obj=None): + """ + Lock every field except ``is_active`` while disabled; owner removal uses + the inline delete action. + """ + fields = super().get_readonly_fields(request, obj) + if obj and not obj.is_active: + editable_fields = [ + f.name + for f in self.model._meta.local_fields + if f.editable and f.name != "is_active" + ] + editable_fields.extend( + f.name for f in self.model._meta.local_many_to_many if f.editable + ) + fields = list(fields) + [f for f in editable_fields if f not in fields] + return fields + + def get_prepopulated_fields(self, request, obj=None): + # Django rejects prepopulated fields that are read-only. + if obj and not obj.is_active: + return {} + return super().get_prepopulated_fields(request, obj) + class Media(CopyableFieldsAdmin.Media): css = {"all": ("openwisp-users/css/admin.css",)} diff --git a/openwisp_users/api/mixins.py b/openwisp_users/api/mixins.py index 0eb32756f..24fe2157f 100644 --- a/openwisp_users/api/mixins.py +++ b/openwisp_users/api/mixins.py @@ -1,6 +1,7 @@ import swapper -from django.core.exceptions import ValidationError +from django.core.exceptions import FieldDoesNotExist, ValidationError from django.db.models import ForeignKey, ManyToManyField, Q +from django.utils.translation import gettext_lazy as _ from django_filters import rest_framework as filters from django_filters.filters import QuerySetRequestMixin as BaseQuerySetRequestMixin from rest_framework.authentication import SessionAuthentication @@ -8,12 +9,22 @@ from rest_framework.permissions import IsAuthenticated from .authentication import BearerAuthentication -from .permissions import DjangoModelPermissions, IsOrganizationManager +from .permissions import ( + DisabledOrgReadOnly, + DjangoModelPermissions, + IsOrganizationManager, +) Organization = swapper.load_model("openwisp_users", "Organization") +DISABLED_ORGANIZATION_ERROR_MESSAGE = _( + 'Organization with pk "{pk_value}" does not exist or is disabled.' +) + class OrgLookup: + select_related_organization = True + @property def org_field(self): return getattr(self, "organization_field", "organization") @@ -22,6 +33,17 @@ def org_field(self): def organization_lookup(self): return f"{self.org_field}__in" + def _organization_relation_is_valid(self, model): + for part in self.org_field.split("__"): + try: + field = model._meta.get_field(part) + except FieldDoesNotExist: + return False + if not (field.concrete and (field.many_to_one or field.one_to_one)): + return False + model = field.related_model + return model == Organization + class SharedObjectsLookup: @property @@ -55,6 +77,10 @@ def queryset_organization_conditions(self): def get_queryset(self): qs = super().get_queryset() + if getattr( + self, "select_related_organization", True + ) and self._organization_relation_is_valid(qs.model): + qs = qs.select_related(self.org_field) if self.request.user.is_superuser: return qs return self.get_organization_queryset(qs) @@ -109,6 +135,10 @@ def assert_parent_exists(self): parent_queryset = self.get_parent_queryset() if not self.request.user.is_superuser: parent_queryset = self.get_organization_queryset(parent_queryset) + if getattr( + self, "select_related_organization", True + ) and self._organization_relation_is_valid(parent_queryset.model): + parent_queryset = parent_queryset.select_related(self.org_field) try: assert parent_queryset.exists() except (AssertionError, ValidationError): @@ -158,29 +188,59 @@ def _user_attr(self): raise NotImplementedError() def filter_fields(self): + """ + Restrict the querysets of writable relational fields so users can + only select active organizations they manage, and related objects + belonging to those organizations (including shared objects when + ``include_shared`` is set). + """ user = self.context["request"].user - # superuser can see everything - if user.is_superuser or user.is_anonymous: + organization_filter = None + if not user.is_superuser and not user.is_anonymous: + organization_filter = getattr(user, self._user_attr) + for name, field in self.fields.items(): + if name == "organization" and not field.read_only: + self._filter_organization_field(field, organization_filter) + else: + self._filter_related_field(field, organization_filter) + + def _filter_organization_field(self, field, organization_filter): + # Keep disabled organizations out of writable relation fields. + queryset = field.queryset.filter(is_active=True) + field.error_messages["does_not_exist"] = DISABLED_ORGANIZATION_ERROR_MESSAGE + view = self.context.get("view") + organization = getattr(self.instance, "organization", None) + if ( + getattr(view, "allow_disabled_organization_writes", False) + and organization is not None + and not organization.is_active + ): + queryset |= field.queryset.filter(pk=organization.pk) + if organization_filter is not None: + field.allow_null = False + allowed_organizations = Q(pk__in=organization_filter) + if ( + getattr(view, "allow_disabled_organization_writes", False) + and organization is not None + and not organization.is_active + ): + allowed_organizations |= Q(pk=organization.pk) + queryset = queryset.filter(allowed_organizations) + field.queryset = queryset + + def _filter_related_field(self, field, organization_filter): + queryset = getattr(field, "queryset", None) + # Read-only and non-relational fields do not expose a queryset. + if queryset is None: return - # non superusers can see only items of organizations they're related to - organization_filter = getattr(user, self._user_attr) - for field in self.fields: - if field == "organization" and not self.fields[field].read_only: - # queryset attribute will not be present if set to read_only - self.fields[field].allow_null = False - self.fields[field].queryset = self.fields[field].queryset.filter( - pk__in=organization_filter - ) - continue - conditions = Q(**{self.organization_lookup: organization_filter}) + conditions = Q(**{f"{self.org_field}__is_active": True}) + if organization_filter is None: + conditions |= Q(**{f"{self.org_field}__isnull": True}) + else: + conditions &= Q(**{self.organization_lookup: organization_filter}) if self.include_shared: - conditions |= Q(organization__isnull=True) - try: - self.fields[field].queryset = self.fields[field].queryset.filter( - conditions - ) - except AttributeError: - pass + conditions |= Q(**{f"{self.org_field}__isnull": True}) + field.queryset = queryset.filter(conditions) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -189,6 +249,11 @@ def __init__(self, *args, **kwargs): if "request" in self.context: self.filter_fields() + def bind(self, field_name, parent): + super().bind(field_name, parent) + if "request" in self.context: + self.filter_fields() + class FilterSerializerByOrgMembership(FilterSerializerByOrganization): """ @@ -308,4 +373,5 @@ class ProtectedAPIMixin(object): permission_classes = ( IsOrganizationManager, DjangoModelPermissions, + DisabledOrgReadOnly, ) diff --git a/openwisp_users/api/permissions.py b/openwisp_users/api/permissions.py index 4177229f3..09d9ee1b4 100644 --- a/openwisp_users/api/permissions.py +++ b/openwisp_users/api/permissions.py @@ -1,5 +1,5 @@ from django.utils.translation import gettext_lazy as _ -from rest_framework.permissions import BasePermission +from rest_framework.permissions import SAFE_METHODS, BasePermission from rest_framework.permissions import ( DjangoModelPermissions as BaseDjangoModelPermissions, ) @@ -38,6 +38,8 @@ def has_object_permission(self, request, view, obj): len(request.user.organizations_managed) >= 1 or len(request.user.organizations_owned) >= 1 ) + if not isinstance(organization, Organization): + return False return self.validate_membership(request.user, organization) def has_permission(self, request, view): @@ -95,6 +97,32 @@ def validate_membership(self, user, org): return org and (user.is_superuser or user.is_owner(org)) +class DisabledOrgReadOnly(ObjectOrganizationMixin, BasePermission): + """ + Keep disabled-organization objects read-only while allowing reads and + deletion. Views can opt out with ``allow_disabled_organization_writes``. + """ + + message = _( + "This object belongs to a disabled organization: " + "it can be viewed or deleted, but not modified." + ) + + def has_object_permission(self, request, view, obj): + if getattr(view, "allow_disabled_organization_writes", False): + return True + if request.method in SAFE_METHODS or request.method == "DELETE": + return True + try: + organization = self.get_object_organization(view, obj) + except AttributeError: + # Do not fail open on a bad relation path; unrelated views opt out. + return False + return organization is None or ( + isinstance(organization, Organization) and organization.is_active + ) + + class DjangoModelPermissions(ObjectOrganizationMixin, BaseDjangoModelPermissions): perms_map = { "GET": ["%(app_label)s.view_%(model_name)s"], diff --git a/openwisp_users/api/serializers.py b/openwisp_users/api/serializers.py index 2328f4971..0f0d15bb3 100644 --- a/openwisp_users/api/serializers.py +++ b/openwisp_users/api/serializers.py @@ -12,6 +12,7 @@ from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.contrib.sites.shortcuts import get_current_site +from django.core.exceptions import ValidationError as DjangoValidationError from django.db import transaction from django.db.models import Q from django.utils.module_loading import import_string @@ -22,6 +23,7 @@ from openwisp_utils.api.serializers import ValidatedModelSerializer from .. import settings as app_settings +from .mixins import DISABLED_ORGANIZATION_ERROR_MESSAGE Group = load_model("openwisp_users", "Group") Organization = load_model("openwisp_users", "Organization") @@ -31,6 +33,19 @@ OrganizationOwner = load_model("openwisp_users", "OrganizationOwner") +def _full_clean_or_raise(instance): + """ + Django's ValidationError raised by full_clean() is not caught by DRF + unless it happens inside a serializer's validate(); these call sites + call full_clean() from create()/update(), so it must be converted + manually or it propagates as an unhandled 500. + """ + try: + instance.full_clean() + except DjangoValidationError as e: + raise serializers.ValidationError(serializers.as_serializer_error(e)) + + class OrganizationSerializer(ValidatedModelSerializer): class Meta: model = Organization @@ -77,7 +92,17 @@ def get_queryset(self): queryset = OrganizationUser.objects.filter( Q(organization__in=user.organizations_managed) ) - return queryset.select_related() + allowed = Q(organization__is_active=True) + organization = getattr(self.root, "instance", None) + if organization is not None and not organization.is_active: + current_org_user_id = ( + OrganizationOwner.objects.filter(organization=organization) + .values_list("organization_user_id", flat=True) + .first() + ) + if current_org_user_id is not None: + allowed |= Q(pk=current_org_user_id) + return queryset.filter(allowed).select_related() class OrganizationOwnerSerializer(serializers.ModelSerializer): @@ -107,6 +132,49 @@ class Meta: "modified", ) + def validate(self, data): + if self.instance and not self.instance.is_active: + owner_data = data.get("owner") or {} + owner_present = "owner" in data + is_owner_unassignment = ( + owner_present and owner_data.get("organization_user") is None + ) + reenabling = data.get("is_active") is True + # Compare values, not submitted keys, so unchanged PUT fields do + # not block re-enabling. + changed_keys = { + key + for key in data + if key != "owner" and getattr(self.instance, key) != data[key] + } + owner_changed = False + if owner_present: + existing_owner = OrganizationOwner.objects.filter( + organization=self.instance + ).first() + existing_org_user = ( + existing_owner.organization_user if existing_owner else None + ) + # Unchanged owners must not block read-modify-write PUTs. + owner_changed = owner_data.get("organization_user") != existing_org_user + if owner_changed: + changed_keys.add("owner") + # Match the admin: while disabled, only re-enable or unassign the owner. + allowed = ( + changed_keys <= {"is_active", "owner"} + and (not owner_changed or is_owner_unassignment) + and (reenabling or is_owner_unassignment) + ) + if not allowed: + raise serializers.ValidationError( + _( + "This organization is disabled: only re-enabling it, " + "unassigning its owner, or deleting it is allowed. Edit " + "other fields or assign an owner after re-enabling it." + ) + ) + return super().validate(data) + def update(self, instance, validated_data): if validated_data.get("owner"): org_owner = validated_data.pop("owner") @@ -118,10 +186,10 @@ def update(self, instance, validated_data): ): org_user = org_owner.get("organization_user") with transaction.atomic(): - org_owner = OrganizationOwner.objects.create( + org_owner = OrganizationOwner( organization=instance, organization_user=org_user ) - org_owner.full_clean() + _full_clean_or_raise(org_owner) org_owner.save() return super().update(instance, validated_data) @@ -135,10 +203,10 @@ def update(self, instance, validated_data): org_user = org_owner.get("organization_user") with transaction.atomic(): existing_owner.first().delete() - org_owner = OrganizationOwner.objects.create( + org_owner = OrganizationOwner( organization=instance, organization_user=org_user ) - org_owner.full_clean() + _full_clean_or_raise(org_owner) org_owner.save() instance = self.instance or self.Meta.model(**validated_data) @@ -185,13 +253,26 @@ def update(self, instance, validated_data): class OrgUserCustomPrimarykeyRelatedField(serializers.PrimaryKeyRelatedField): + default_error_messages = { + "does_not_exist": DISABLED_ORGANIZATION_ERROR_MESSAGE, + } + def get_queryset(self): user = self.context["request"].user if user.is_superuser: queryset = Organization.objects.all() else: queryset = Organization.objects.filter(pk__in=user.organizations_managed) - return queryset + allowed = Q(is_active=True) + # Existing disabled memberships must resolve so the deletion path works. + target_user = getattr(self.root, "instance", None) + if target_user is not None: + # Keep this as a subquery to avoid an extra round trip. + existing_disabled_orgs = OrganizationUser.objects.filter( + user=target_user, organization__is_active=False + ).values("organization_id") + allowed |= Q(pk__in=existing_disabled_orgs) + return queryset.filter(allowed) class OrganizationUserSerializer(serializers.ModelSerializer): @@ -281,20 +362,22 @@ def create(self, validated_data): password = validated_data.pop("password") email_verified = validated_data.pop("email_verified", False) - instance = self.instance or self.Meta.model(**validated_data) - instance.set_password(password) - instance.full_clean() - instance.save() + # Roll back the user if membership validation fails. + with transaction.atomic(): + instance = self.instance or self.Meta.model(**validated_data) + instance.set_password(password) + _full_clean_or_raise(instance) + instance.save() - if group_data: - instance.groups.add(*group_data) + if group_data: + instance.groups.add(*group_data) - if org_user_data: - if org_user_data.get("organization") is not None: - org_user_data["user"] = instance - org_user_instance = OrganizationUser(**org_user_data) - org_user_instance.full_clean() - org_user_instance.save() + if org_user_data: + if org_user_data.get("organization") is not None: + org_user_data["user"] = instance + org_user_instance = OrganizationUser(**org_user_data) + _full_clean_or_raise(org_user_instance) + org_user_instance.save() if instance.email: try: @@ -365,20 +448,21 @@ def update(self, instance, validated_data): except OrganizationUser.DoesNotExist: pass if org_user: - if ( - str(org_user_data["organization"].id) - in instance.organizations_dict.keys() - ): - if org_user.is_admin != org_user_data.get("is_admin"): + # Explicit contract for an existing membership: + # - is_admin omitted: leave the membership unchanged + # - is_admin sent and changed: update it; + # - is_admin sent unchanged -> remove the membership + if "is_admin" in org_user_data: + if org_user.is_admin != org_user_data["is_admin"]: org_user.is_admin = org_user_data["is_admin"] - org_user.full_clean() + _full_clean_or_raise(org_user) org_user.save() else: org_user.delete() else: org_user_data["user"] = instance org_user_instance = OrganizationUser(**org_user_data) - org_user_instance.full_clean() + _full_clean_or_raise(org_user_instance) org_user_instance.save() return super().update(instance, validated_data) diff --git a/openwisp_users/apps.py b/openwisp_users/apps.py index 3c70ba060..dd5a2be80 100644 --- a/openwisp_users/apps.py +++ b/openwisp_users/apps.py @@ -16,6 +16,7 @@ from . import settings as app_settings from .auth import SESAME_BACKEND, record_password_based_login +from .signals import organization_disabled, organization_enabled logger = logging.getLogger(__name__) @@ -108,11 +109,15 @@ def connect_receivers(self): (post_delete, "post_delete"), ] - pre_save.connect( - self.handle_org_is_active_change, - sender=Organization, - dispatch_uid="handle_org_is_active_change", - ) + for signal, name in ( + (organization_disabled, "organization_disabled"), + (organization_enabled, "organization_enabled"), + ): + signal.connect( + self.handle_org_is_active_change, + sender=Organization, + dispatch_uid=f"handle_org_is_active_change_{name}", + ) for model in [OrganizationUser, OrganizationOwner]: for signal, name in signal_tuples: @@ -175,18 +180,9 @@ def handle_allauth_login(cls, request, sociallogin=None, **kwargs): @classmethod def handle_org_is_active_change(cls, instance, **kwargs): - if instance._state.adding: - # If it's a new organization, we don't need to update any cache - return - Organization = instance._meta.model - try: - old_instance = Organization.objects.only("is_active").get(pk=instance.pk) - except Organization.DoesNotExist: - return from .tasks import invalidate_org_membership_cache - if instance.is_active != old_instance.is_active: - invalidate_org_membership_cache.delay(instance.pk) + invalidate_org_membership_cache.delay(instance.pk) @classmethod def pre_save_update_organizations_dict(cls, instance, **kwargs): @@ -232,7 +228,7 @@ def update_organizations_dict(cls, instance, signal, **kwargs): @classmethod def create_organization_owner(cls, instance, created, **kwargs): - if not created or not instance.is_admin: + if not created or not instance.is_admin or not instance.organization.is_active: return OrganizationOwner = load_model("openwisp_users", "OrganizationOwner") org_owner_exist = OrganizationOwner.objects.filter( diff --git a/openwisp_users/base/models.py b/openwisp_users/base/models.py index f8471cc10..0a551292d 100644 --- a/openwisp_users/base/models.py +++ b/openwisp_users/base/models.py @@ -1,3 +1,4 @@ +import copy import logging import uuid from smtplib import SMTPException @@ -22,6 +23,7 @@ from openwisp_utils.admin_theme.email import send_email from .. import settings as app_settings +from ..signals import organization_disabled, organization_enabled from ..utils import throttle_email_batch logger = logging.getLogger(__name__) @@ -479,6 +481,28 @@ def __str__(self): class Meta: abstract = True + def save(self, *args, **kwargs): + is_new = self._state.adding + update_fields = kwargs.get("update_fields") + previous_is_active = None + if not is_new and (update_fields is None or "is_active" in update_fields): + previous_is_active = ( + self.__class__.objects.filter(pk=self.pk) + .values_list("is_active", flat=True) + .first() + ) + super().save(*args, **kwargs) + if ( + not is_new + and (update_fields is None or "is_active" in update_fields) + and self.is_active != previous_is_active + ): + signal = organization_enabled if self.is_active else organization_disabled + instance = copy.copy(self) + transaction.on_commit( + lambda: signal.send(sender=self.__class__, instance=instance) + ) + def add_user(self, user, is_admin=False, **kwargs): """ We override this method from the upstream dependency to @@ -486,14 +510,16 @@ def add_user(self, user, is_admin=False, **kwargs): automatically via a signal receiver. Without this change, the add_user method would throw IntegrityError. """ - if not self.users.all().exists(): is_admin = True OrganizationUser = load_model("openwisp_users", "OrganizationUser") - return OrganizationUser.objects.create( - user=user, organization=self, is_admin=is_admin + org_user = OrganizationUser( + user=user, organization=self, is_admin=is_admin, **kwargs ) + org_user.full_clean() + org_user.save() + return org_user class BaseOrganizationUser(models.Model): @@ -508,6 +534,38 @@ class Meta: abstract = True def clean(self): + original = None + if not self._state.adding: + original = ( + self.__class__.objects.select_related("organization") + .filter(pk=self.pk) + .first() + ) + + changed = original is None or any( + getattr(original, field.attname) != getattr(self, field.attname) + for field in self._meta.concrete_fields + if field.editable and not field.primary_key + ) + + if changed and self.organization_id is not None: + if not self.organization.is_active: + if self._state.adding: + raise ValidationError( + _("Cannot add users to a disabled organization.") + ) + raise ValidationError( + _("Memberships of a disabled organization cannot be modified.") + ) + if ( + original + and original.organization_id != self.organization_id + and not original.organization.is_active + ): + raise ValidationError( + _("Memberships of a disabled organization cannot be modified.") + ) + if ( not self._state.adding and self.user.is_owner(self.organization_id) @@ -538,7 +596,30 @@ class BaseOrganizationOwner(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) def clean(self): - if self.organization_user.organization.pk != self.organization.pk: + original = None + if self.pk: + original = ( + self.__class__.objects.select_related("organization") + .filter(pk=self.pk) + .first() + ) + changed = ( + original is None + or original.organization_id != self.organization_id + or original.organization_user_id != self.organization_user_id + ) + if changed and ( + not self.organization.is_active + or ( + original + and original.organization_id != self.organization_id + and not original.organization.is_active + ) + ): + raise ValidationError( + _("Cannot assign an owner to a disabled organization.") + ) + if self.organization_user.organization_id != self.organization_id: raise ValidationError( { "organization_user": _( diff --git a/openwisp_users/multitenancy.py b/openwisp_users/multitenancy.py index b6557fb23..28046b2bc 100644 --- a/openwisp_users/multitenancy.py +++ b/openwisp_users/multitenancy.py @@ -1,5 +1,7 @@ +from django.contrib import messages from django.contrib.auth import get_user_model from django.db.models import Q +from django.http import HttpResponseRedirect from django.utils.translation import gettext_lazy as _ from swapper import load_model @@ -20,6 +22,9 @@ class MultitenantAdminMixin(object): multitenant_shared_relations = None multitenant_parent = None + # Set False on subclasses that allow writes to disabled-organization objects. + disabled_organization_write_protection = True + disabled_organization_action_exclusions = () def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -47,30 +52,139 @@ def get_queryset(self, request): if self.model == User: return self.multitenant_behaviour_for_user_admin(request) if user.is_superuser: + # Autocomplete requests exclude objects associated with + # disabled organizations. + if "field_name" in request.GET and hasattr(self.model, "organization"): + active_or_shared = Q(organization__is_active=True) | Q( + organization=None + ) + return qs.filter(active_or_shared) return qs if hasattr(self.model, "organization"): - return qs.filter(organization__in=user.organizations_managed) + return qs.filter( + organization__in=user.organizations_managed, + organization__is_active=True, + ) if self.model.__name__ == "Organization": - return qs.filter(pk__in=user.organizations_managed) + return qs.filter(pk__in=user.organizations_managed, is_active=True) elif not self.multitenant_parent: return qs else: qsarg = "{0}__organization__in".format(self.multitenant_parent) - return qs.filter(**{qsarg: user.organizations_managed}) + active_qsarg = "{0}__organization__is_active".format( + self.multitenant_parent + ) + return qs.filter(**{qsarg: user.organizations_managed, active_qsarg: True}) + + def get_object_organization(self, obj): + """ + Resolve an object's organization, including through + ``multitenant_parent``. + """ + if self.model.__name__ == "Organization": + return obj + organization = getattr(obj, "organization", None) + if organization is None and self.multitenant_parent: + parent = obj + for attr in self.multitenant_parent.split("__"): + parent = getattr(parent, attr, None) + if parent is None: + break + organization = getattr(parent, "organization", None) + return organization - def _edit_form(self, request, form): + def has_change_permission(self, request, obj=None): """ - Modifies the form querysets as follows; - if current user is not superuser: - * show only relevant organizations - * show only relations associated to relevant organizations - or shared relations - * do not allow organization field to be empty (shared org) - else show everything + Block changes to disabled organizations unless the admin opts out. + """ + if self.disabled_organization_write_protection and obj is not None: + organization = self.get_object_organization(obj) + if organization is not None and not organization.is_active: + return False + return super().has_change_permission(request, obj) + + def get_inline_instances(self, request, obj=None): + """ + Disable add/change for inlines on objects from disabled organizations + while keeping delete. + """ + inlines = super().get_inline_instances(request, obj) + if obj is None or not self.disabled_organization_write_protection: + return inlines + organization = self.get_object_organization(obj) + if organization is None or organization.is_active: + return inlines + for inline in inlines: + if getattr(inline, "disabled_organization_write_protection", True): + inline.has_add_permission = lambda request, obj=None: False + inline.has_change_permission = lambda request, obj=None: False + return inlines + + def has_add_permission(self, request, *args, **kwargs): + """ + Hide unusable add forms when no active organization is managed. + """ + if ( + not request.user.is_superuser + and self.model != User + and not request.user.organizations_managed + ): + # The form requires an organization, so it cannot work without one. + if hasattr(self.model, "organization") or self.multitenant_parent: + return False + return super().has_add_permission(request, *args, **kwargs) + + def response_action(self, request, queryset): + action = request.POST.get("action") + if ( + self.disabled_organization_write_protection + and action not in self.get_disabled_organization_action_exclusions() + and any( + organization is not None and not organization.is_active + for organization in ( + self.get_object_organization(obj) for obj in queryset + ) + ) + ): + self.message_user( + request, + _("Actions cannot modify objects of disabled organizations."), + messages.ERROR, + ) + return HttpResponseRedirect(request.get_full_path()) + return super().response_action(request, queryset) + + def get_disabled_organization_action_exclusions(self): + return {"delete_selected", "delete_selected_overridden"}.union( + self.disabled_organization_action_exclusions + ) + + def _edit_form(self, request, form, obj=None): + """ + Filter form fields by organization and exclude disabled choices. + + An opted-out admin keeps the object's current disabled organization + selectable so the existing object can still be saved. """ fields = form.base_fields user = request.user org_field = fields.get("organization") + keep_disabled_org_pk = None + if not self.disabled_organization_write_protection and obj is not None: + organization = self.get_object_organization(obj) + if organization is not None and not organization.is_active: + keep_disabled_org_pk = organization.pk + if org_field: + allowed = Q(is_active=True) + if keep_disabled_org_pk is not None: + allowed |= Q(pk=keep_disabled_org_pk) + org_field.queryset = org_field.queryset.filter(allowed) + active_or_shared = Q(organization__is_active=True) | Q(organization=None) + for field_name in self.multitenant_shared_relations: + if field_name not in fields: + continue + field = fields[field_name] + field.queryset = field.queryset.filter(active_or_shared) if user.is_superuser and org_field and not org_field.required: org_field.empty_label = SHARED_SYSTEMWIDE_LABEL elif not user.is_superuser: @@ -78,14 +192,14 @@ def _edit_form(self, request, form): # organizations relation; # may be readonly and not present in field list if org_field: - org_field.queryset = org_field.queryset.filter(pk__in=orgs_pk) + managed = Q(pk__in=orgs_pk) + if keep_disabled_org_pk is not None: + managed |= Q(pk=keep_disabled_org_pk) + org_field.queryset = org_field.queryset.filter(managed) org_field.empty_label = None org_field.required = True - # other relations q = Q(organization__in=orgs_pk) | Q(organization=None) for field_name in self.multitenant_shared_relations: - # each relation may be readonly - # and not present in field list if field_name not in fields: continue field = fields[field_name] @@ -93,11 +207,11 @@ def _edit_form(self, request, form): def get_form(self, request, obj=None, **kwargs): form = super().get_form(request, obj, **kwargs) - self._edit_form(request, form) + self._edit_form(request, form, obj) return form def get_formset(self, request, obj=None, **kwargs): - formset = super().get_formset(request, obj=None, **kwargs) + formset = super().get_formset(request, obj, **kwargs) self._edit_form(request, formset.form) return formset diff --git a/openwisp_users/signals.py b/openwisp_users/signals.py new file mode 100644 index 000000000..48dbc8985 --- /dev/null +++ b/openwisp_users/signals.py @@ -0,0 +1,10 @@ +from django.dispatch import Signal + +organization_disabled = Signal() +organization_disabled.__doc__ = """ +Providing arguments: ['instance'] +""" +organization_enabled = Signal() +organization_enabled.__doc__ = """ +Providing arguments: ['instance'] +""" diff --git a/openwisp_users/static/openwisp-users/js/org-autocomplete.js b/openwisp_users/static/openwisp-users/js/org-autocomplete.js index 08b78c7b4..278a8e432 100644 --- a/openwisp_users/static/openwisp-users/js/org-autocomplete.js +++ b/openwisp_users/static/openwisp-users/js/org-autocomplete.js @@ -7,58 +7,61 @@ // Therefore, the backend uses "null" id for Systemwide shared // objects. This causes issues on submitting forms because // Django expects an empty string (for None) or a UUID string. - // Hence, we need to update the value of selected option before - // submission of form. - var formElement = $("select#id_organization"); - while (formElement.prop("tagName") !== "FORM") { - formElement = formElement.parent(); - } - formElement.submit(function () { - var target = $("select#id_organization option:selected"); - if (target.val() === "null") { - target.val(""); - } + // Hence, we need to update the value of the selected option before + // submission of the form. + // + // Find the organization fields at submit time so dynamically added + // inlines are included too. + $("form").on("submit", function () { + $(this) + .find("select[data-field-name='organization'] option:selected") + .each(function () { + var selected = $(this); + if (selected.val() === "null") { + selected.val(""); + } + }); }); - if (!$("select#id_organization").val()) { - var orgField = $("#id_organization"), - pathName = window.location.pathname.split("/"); - // If the field is rendered empty on a change form, then the - // the object is shared systemwide (no organization). - if (pathName[pathName.length - 2] == "change") { - orgField.val("null"); - orgField.trigger("change"); - return; - } + // Auto-selection only applies to the single top-level organization field + // when it is still empty (e.g. add forms or systemwide-shared objects). Skip + // inline organization selects and forms where an organization is already set. + var orgField = $("select#id_organization"); + if (!orgField.length || orgField.val()) { + return; + } - // If only one organization option is available, then select that - // organization automatically - $.ajax({ - url: orgField.data("ajax--url"), - data: { - app_label: orgField.data("app-label"), - model_name: orgField.data("model-name"), - field_name: orgField.data("field-name"), - }, - success: function (data) { - if (data.results.length === 1) { - var option = new Option( - data.results[0].text, - data.results[0].id, - true, - true, - ); - orgField.append(option).trigger("change"); - // manually trigger the `select2:select` event - orgField.trigger({ - type: "select2:select", - params: { - data: data.results[0], - }, - }); - } - }, - }); + var pathName = window.location.pathname.split("/"); + // If the field is rendered empty on a change form, then the + // object is shared systemwide (no organization). + if (pathName[pathName.length - 2] == "change") { + orgField.val("null"); + orgField.trigger("change"); + return; } + + // If only one organization option is available, then select that + // organization automatically. + $.ajax({ + url: orgField.data("ajax--url"), + data: { + app_label: orgField.data("app-label"), + model_name: orgField.data("model-name"), + field_name: orgField.data("field-name"), + }, + success: function (data) { + if (data.results.length === 1) { + var option = new Option(data.results[0].text, data.results[0].id, true, true); + orgField.append(option).trigger("change"); + // manually trigger the `select2:select` event + orgField.trigger({ + type: "select2:select", + params: { + data: data.results[0], + }, + }); + } + }, + }); }); })(django.jQuery); diff --git a/openwisp_users/tests/test_admin.py b/openwisp_users/tests/test_admin.py index 1a2eb331c..2b2fa9ef8 100644 --- a/openwisp_users/tests/test_admin.py +++ b/openwisp_users/tests/test_admin.py @@ -13,7 +13,7 @@ from django.core.exceptions import ValidationError from django.db import DEFAULT_DB_ALIAS from django.template.defaultfilters import date -from django.test import TestCase, override_settings +from django.test import RequestFactory, TestCase, override_settings from django.urls import reverse from django.utils.timezone import localdate, now, timedelta from freezegun import freeze_time @@ -27,7 +27,9 @@ from ..apps import logger as apps_logger from ..auth import SESSION_KEY from ..multitenancy import MultitenantAdminMixin +from ..widgets import OrganizationAutocompleteSelect from .utils import ( + TestDisabledOrgAdminMixin, TestMultitenantAdminMixin, TestOrganizationMixin, TestUserAdditionalFieldsMixin, @@ -42,7 +44,7 @@ class TestUsersAdmin( AdminActionPermTestMixin, - TestOrganizationMixin, + TestDisabledOrgAdminMixin, TestUserAdditionalFieldsMixin, TestCase, ): @@ -98,6 +100,9 @@ def _get_api_key_inline_params(self, user, generate_token=False): params.update({"auth_token-0-generate_token": "on"}) return params + def _get_disabled_org_test_excluded_inline(self): + return [] + @property def add_user_inline_params(self): params = { @@ -1419,8 +1424,7 @@ def test_action_active_clears_expired_expiration_date(self): self.assertEqual(user.is_active, True) if expected_expiration_date is None: expected_message = ( - "Successfully activated 1 user and cleared 1 expiration" - " date." + "Successfully activated 1 user and cleared 1 expiration date." ) else: expected_message = "Successfully activated 1 user." @@ -1884,6 +1888,314 @@ def test_only_superuser_can_delete_inline_org_owner(self): self.assertEqual(r.status_code, 200) self.assertContains(r, '-DELETE">Delete') + def test_disabled_organization_change_form(self): + admin = self._get_admin() + self.client.force_login(admin) + org = self._create_org(name="disabled-admin-org", is_active=False) + path = reverse(f"admin:{self.app_label}_organization_change", args=[org.pk]) + + with self.subTest("Fields readonly except is_active, no 500"): + response = self.client.get(path) + self.assertEqual(response.status_code, 200) + self.assertNotContains( + response, f' count_before, created) + + def _test_disabled_org_admin_org_field_excludes_disabled( + self, + url, + disabled_org, + roles=("superuser",), + organization=None, + role_kwargs=None, + ): + role_kwargs = role_kwargs or {} + for role in roles: + with self.subTest(role=role): + user = self._disabled_org_role_user( + role, organization=organization, **role_kwargs.get(role, {}) + ) + self.client.force_login(user) + response = self.client.get(url) + self.assertNotContains(response, f"{disabled_org.name}") + self.client.logout() + + def _test_disabled_org_admin_crud( + self, + obj, + change_data, + roles=("org_admin", "superuser"), + operations=("view", "change", "delete", "add"), + organization=None, + org_admin_expected=None, + superuser_expected=None, + unchanged_field="name", + create_data=None, + ): + """Run shared checks for direct or parent-linked organizations.""" + if create_data is None: + operations = tuple(op for op in operations if op != "add") + organization = organization or getattr(obj, "organization", None) + urls = self._get_disabled_org_admin_urls(obj) + specs = { + "org_admin": { + **self.disabled_org_admin_default_expectations["org_admin"], + **(org_admin_expected or {}), + }, + "superuser": { + **self.disabled_org_admin_default_expectations["superuser"], + **(superuser_expected or {}), + }, + } + for role in roles: + user = self._disabled_org_role_user(role, organization=organization) + self.client.force_login(user) + for operation in operations: + spec = specs[role][operation] + with self.subTest(role=role, operation=operation): + if operation == "view": + self._test_disabled_org_admin_view(urls["view"], **spec) + elif operation == "change": + self._test_disabled_org_admin_change( + urls["change"], + change_data, + obj, + unchanged_field=unchanged_field, + **spec, + ) + elif operation == "delete": + self._test_disabled_org_admin_delete( + urls["delete"], type(obj), obj.pk, **spec + ) + elif operation == "add": + self._test_disabled_org_admin_add( + urls["add"], create_data, type(obj), **spec + ) + else: + raise ValueError(f"Unknown operation: {operation!r}") + self.client.logout() + + def _test_disabled_org_admin_inline_readonly( + self, + model_admin, + disabled_obj, + active_obj=None, + inline_models=None, + inline_admins=None, + user=None, + ): + request = RequestFactory().get("/") + request.user = user or self._get_admin() + + inlines = inline_admins or model_admin.get_inline_instances( + request, disabled_obj + ) + if inline_models is not None: + inlines = [i for i in inlines if isinstance(i, inline_models)] + self.assertNotEqual(inlines, []) + for inline in inlines: + self.assertEqual(inline.has_add_permission(request, disabled_obj), False) + self.assertEqual(inline.has_change_permission(request, disabled_obj), False) + self.assertEqual(inline.has_delete_permission(request, disabled_obj), True) + + if active_obj is not None: + active_inlines = model_admin.get_inline_instances(request, active_obj) + if inline_models is not None: + active_inlines = [ + i for i in active_inlines if isinstance(i, inline_models) + ] + for inline in active_inlines: + self.assertEqual( + inline.has_change_permission(request, active_obj), True + ) + + +class TestMultitenantAdminMixin(TestDisabledOrgAdminMixin): def setUp(self): admin = self._create_admin(password="tester") admin.organizations_dict # force caching + super().setUp() def _login(self, username="admin", password="tester"): self.client.login(username=username, password=password) @@ -182,15 +388,24 @@ def _logout(self): self.client.logout() def _test_multitenant_admin( - self, url, visible, hidden, select_widget=False, administrator=False + self, + url, + visible, + hidden, + select_widget=False, + administrator=False, + superuser_hidden=None, ): """ reusable test function that ensures different users can see the right objects. an operator with limited permissions will not be able to see the elements contained in ``hidden``, while - a superuser can see everything. + a superuser can see everything, except the elements in + ``superuser_hidden`` (e.g. objects belonging to a disabled + organization, which relation pickers exclude for everyone). """ + superuser_hidden = superuser_hidden or [] if administrator: self._login(username="administrator", password="tester") else: @@ -222,12 +437,16 @@ def _f(el, select_widget=False): self._logout() self._login(username="admin", password="tester") response = self.client.get(url) - # ensure all elements are visible to superuser - all_elements = visible + hidden + # Relation pickers still hide disabled values from superusers. + all_elements = [el for el in visible + hidden if el not in superuser_hidden] for el in all_elements: self.assertContains( response, _f(el, select_widget), msg_prefix="[superuser contains]" ) + for el in superuser_hidden: + self.assertNotContains( + response, _f(el, select_widget), msg_prefix="[superuser not-contains]" + ) def _test_recoverlist_operator_403(self, app_label, model_label): self._login(username="operator", password="tester") diff --git a/openwisp_users/views.py b/openwisp_users/views.py index 5fbb9c757..95f3d9e29 100644 --- a/openwisp_users/views.py +++ b/openwisp_users/views.py @@ -33,7 +33,12 @@ def get_queryset(self): qs = super().get_queryset() org_lookup = self.get_org_lookup() if not self.request.user.is_superuser and org_lookup: - return qs.filter(**{org_lookup: self.request.user.organizations_managed}) + qs = qs.filter(**{org_lookup: self.request.user.organizations_managed}) + if ( + qs.model == Organization + and self.request.GET.get("exclude_disabled") == "true" + ): + qs = qs.filter(is_active=True) return qs def get_empty_label(self): diff --git a/openwisp_users/widgets.py b/openwisp_users/widgets.py index 3b23fed67..9c8daf46c 100644 --- a/openwisp_users/widgets.py +++ b/openwisp_users/widgets.py @@ -10,7 +10,7 @@ class Media: js = ["admin/js/jquery.init.js", "openwisp-users/js/org-autocomplete.js"] def get_url(self): - return reverse("admin:ow-auto-filter") + return reverse("admin:ow-auto-filter") + "?exclude_disabled=true" def optgroups(self, name, value, attrs=None): groups = super().optgroups(name, value, attrs) diff --git a/tests/testapp/__init__.py b/tests/testapp/__init__.py index 823fc7517..8b4d09742 100644 --- a/tests/testapp/__init__.py +++ b/tests/testapp/__init__.py @@ -38,3 +38,11 @@ def _create_tag(self, **kwargs): tag.full_clean() tag.save() return tag + + def _create_config(self, **kwargs): + options = dict(name="test-config") + options.update(kwargs) + config = self.config_model(**options) + config.full_clean() + config.save() + return config diff --git a/tests/testapp/admin.py b/tests/testapp/admin.py index e565f6486..18095f685 100644 --- a/tests/testapp/admin.py +++ b/tests/testapp/admin.py @@ -1,25 +1,34 @@ from django.contrib import admin from django.utils.translation import gettext_lazy as _ +from openwisp_users.admin import MultitenantReadOnlyInlineFormSet, RequiredInlineFormSet from openwisp_users.multitenancy import ( MultitenantAdminMixin, MultitenantOrgFilter, MultitenantRelatedOrgFilter, ) -from .models import Book, Library, Shelf, Tag, Template +from .models import Bio, Book, Bookmark, Config, Library, Shelf, Tag, Template class BaseAdmin(MultitenantAdminMixin, admin.ModelAdmin): pass +class BookInline(admin.TabularInline): + # Verify the parent mixin protects inlines that do not use it. + model = Book + fields = ["name", "author"] + extra = 0 + + class ShelfAdmin(BaseAdmin): list_display = ["name", "organization"] list_filter = [MultitenantOrgFilter] fields = ["name", "organization", "tags", "created", "modified"] search_fields = ["name"] multitenant_shared_relations = ["tags"] + inlines = [BookInline] class ShelfFilter(MultitenantRelatedOrgFilter): @@ -35,6 +44,7 @@ class BookAdmin(BaseAdmin): ShelfFilter, ] fields = ["name", "author", "organization", "shelf", "created", "modified"] + autocomplete_fields = ["shelf"] multitenant_shared_relations = ["shelf"] def change_view(self, request, object_id, form_url="", extra_context=None): @@ -68,8 +78,44 @@ class TagAdmin(BaseAdmin): pass +class LibraryParentAdmin(MultitenantAdminMixin, admin.ModelAdmin): + # Resolve the organization through Book for parent traversal coverage. + multitenant_parent = "book" + + +class ConfigAdmin(BaseAdmin): + # Exercise the write-protection opt-out through the admin endpoints. + disabled_organization_write_protection = False + fields = ["name", "organization", "template"] + + +class BioInlineFormSet(MultitenantReadOnlyInlineFormSet, RequiredInlineFormSet): + pass + + +class BioInline(MultitenantAdminMixin, admin.StackedInline): + model = Bio + formset = BioInlineFormSet + fields = ["website", "organization"] + extra = 0 + + +class BookmarkInlineFormSet(MultitenantReadOnlyInlineFormSet): + organization_fk_field = "book" + organization_lookup = "book__organization" + + +class BookmarkInline(MultitenantAdminMixin, admin.StackedInline): + model = Bookmark + formset = BookmarkInlineFormSet + fields = ("book",) + extra = 0 + multitenant_shared_relations = ["book"] + + admin.site.register(Shelf, ShelfAdmin) admin.site.register(Book, BookAdmin) admin.site.register(Template, TemplateAdmin) -admin.site.register(Library) +admin.site.register(Library, LibraryParentAdmin) admin.site.register(Tag, TagAdmin) +admin.site.register(Config, ConfigAdmin) diff --git a/tests/testapp/migrations/0007_bio.py b/tests/testapp/migrations/0007_bio.py new file mode 100644 index 000000000..410790fba --- /dev/null +++ b/tests/testapp/migrations/0007_bio.py @@ -0,0 +1,81 @@ +# Generated by Django 5.2.16 on 2026-08-21 09:55 + +import django.db.models.deletion +import swapper +from django.conf import settings +from django.db import migrations, models + +import openwisp_users.mixins + + +class Migration(migrations.Migration): + dependencies = [ + swapper.dependency("openwisp_users", "Organization"), + ("testapp", "0006_alter_book_shelf"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="Bio", + fields=[ + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("website", models.URLField(blank=True, verbose_name="website")), + ( + "organization", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to=swapper.get_model_name("openwisp_users", "Organization"), + verbose_name="organization", + ), + ), + ( + "user", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="bios", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={"abstract": False}, + bases=(openwisp_users.mixins.ValidateOrgMixin, models.Model), + ), + migrations.CreateModel( + name="Bookmark", + fields=[ + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "book", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to="testapp.book", + ), + ), + ( + "user", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="bookmarks", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + ), + ] diff --git a/tests/testapp/models.py b/tests/testapp/models.py index b43c834fe..60b90f142 100644 --- a/tests/testapp/models.py +++ b/tests/testapp/models.py @@ -1,3 +1,4 @@ +from django.conf import settings from django.core.exceptions import ValidationError from django.db import models from django.utils.translation import gettext_lazy as _ @@ -70,3 +71,20 @@ class Library(models.Model): def __str__(self): return self.name + + +class Bio(OrgMixin): + user = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="bios" + ) + website = models.URLField(_("website"), blank=True) + + def __str__(self): + return self.website + + +class Bookmark(models.Model): + user = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="bookmarks" + ) + book = models.ForeignKey(Book, on_delete=models.CASCADE) diff --git a/tests/testapp/tests/mixins.py b/tests/testapp/tests/mixins.py index 4f323cd49..078c469d1 100644 --- a/tests/testapp/tests/mixins.py +++ b/tests/testapp/tests/mixins.py @@ -1,10 +1,10 @@ -from openwisp_users.tests.test_api import AuthenticationMixin +from openwisp_users.tests.test_api import AuthenticationMixin, TestDisabledOrgApiMixin from openwisp_users.tests.utils import TestMultitenantAdminMixin from .. import CreateMixin class TestMultitenancyMixin( - CreateMixin, TestMultitenantAdminMixin, AuthenticationMixin + CreateMixin, TestMultitenantAdminMixin, TestDisabledOrgApiMixin, AuthenticationMixin ): pass diff --git a/tests/testapp/tests/test_admin.py b/tests/testapp/tests/test_admin.py index b27b3a337..b33d560ef 100644 --- a/tests/testapp/tests/test_admin.py +++ b/tests/testapp/tests/test_admin.py @@ -1,14 +1,16 @@ import os import django +from django.contrib import admin from django.contrib.auth import get_user_model -from django.test import TestCase +from django.test import RequestFactory, TestCase from django.urls import reverse from swapper import load_model from openwisp_users.tests.utils import TestOrganizationMixin -from ..models import Template +from ..admin import BookmarkInline +from ..models import Book, Bookmark, Template Organization = load_model("openwisp_users", "Organization") OrganizationUser = load_model("openwisp_users", "OrganizationUser") @@ -44,10 +46,36 @@ def test_accounts_login(self): r, '', html=True ) + def test_indirect_organization_inline_readonly_for_disabled_org(self): + admin_user = self._create_admin() + organization = self._create_org(name="disabled-bookmark-org", is_active=False) + user = self._create_user( + username="disabled-bookmark-user", + email="disabled-bookmark-user@example.com", + ) + book = Book.objects.create( + name="Disabled organization book", + author="Test author", + organization=organization, + ) + bookmark = Bookmark.objects.create(user=user, book=book) + request = RequestFactory().get( + reverse(f"admin:{self.app_label}_user_change", args=[user.pk]) + ) + request.user = admin_user + inline = BookmarkInline(User, admin.site) + formset_class = inline.get_formset(request, user) + formset = formset_class(instance=user, prefix="bookmarks") + form = formset.forms[0] + self.assertEqual(form.instance.pk, bookmark.pk) + self.assertEqual(form.fields["book"].disabled, True) + self.assertEqual(form.fields["book"].queryset.filter(pk=book.pk).exists(), True) + class TestTemplateAdmin(TestOrganizationMixin, TestCase): def test_org_admin_create_shareable_template(self): - administrator = self._create_administrator() + org = self._create_org(name="test-org") + administrator = self._create_administrator(organizations=[org]) self.client.force_login(administrator) response = self.client.post( reverse("admin:testapp_template_add"), diff --git a/tests/testapp/tests/test_filter_classes.py b/tests/testapp/tests/test_filter_classes.py index 5d5698659..da754834c 100644 --- a/tests/testapp/tests/test_filter_classes.py +++ b/tests/testapp/tests/test_filter_classes.py @@ -309,6 +309,30 @@ def test_post_book_nested_shelf(self): self.assertEqual(Shelf.objects.count(), 3) self.assertEqual(Book.objects.count(), 3) + def test_post_book_nested_shelf_rejects_disabled_organization(self): + org = self._get_org("org_a") + disabled_org = self._create_org(name="disabled-org", is_active=False) + administrator = self._create_administrator() + self._create_org_user(user=administrator, is_admin=True, organization=org) + token = self._obtain_auth_token(administrator) + response = self.client.post( + reverse("test_book_nested_shelf"), + { + "shelf": { + "name": "disabled-shelf", + "organization": disabled_org.pk, + }, + "name": "disabled-book", + "author": "test-author", + "organization": org.pk, + }, + content_type="application/json", + HTTP_AUTHORIZATION=f"Bearer {token}", + ) + self.assertEqual(response.status_code, 400) + self.assertEqual(Shelf.objects.filter(name="disabled-shelf").exists(), False) + self.assertEqual(Book.objects.filter(name="disabled-book").exists(), False) + def test_shelf_with_read_only_org_field(self): org1 = self._create_org(name="org1") operator = self._get_operator() diff --git a/tests/testapp/tests/test_multitenancy.py b/tests/testapp/tests/test_multitenancy.py index 6a704df4c..aaab31020 100644 --- a/tests/testapp/tests/test_multitenancy.py +++ b/tests/testapp/tests/test_multitenancy.py @@ -1,13 +1,31 @@ -from django.test import TestCase +from django.contrib import admin +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Permission +from django.contrib.messages.storage.cookie import CookieStorage +from django.test import RequestFactory, TestCase from django.urls import reverse -from ..models import Book, Shelf +from openwisp_users.multitenancy import MultitenantAdminMixin + +from ..admin import BookInline, LibraryParentAdmin, ShelfAdmin +from ..models import Book, Config, Library, Shelf from .mixins import TestMultitenancyMixin +User = get_user_model() + + +class ShelfDisabledOrgWriteAllowedAdmin(MultitenantAdminMixin, admin.ModelAdmin): + # Test the opt-out separately so ShelfAdmin's default remains covered. + disabled_organization_write_protection = False + fields = ["name", "organization"] + inlines = [BookInline] + class TestMultitenancy(TestMultitenancyMixin, TestCase): book_model = Book shelf_model = Shelf + library_model = Library + config_model = Config def _create_multitenancy_test_env(self): org1 = self._create_org(name="org1") @@ -59,11 +77,287 @@ def test_book_queryset(self): ) def test_book_shelf_fk_queryset(self): + data = self._create_multitenancy_test_env() + url = reverse("admin:testapp_book_add") + cases = ( + ("administrator", {data["s1"].pk}), + ("admin", {data["s1"].pk, data["s2"].pk}), + ) + for username, expected_shelf_pks in cases: + with self.subTest(username=username): + self._login(username=username, password="tester") + response = self.client.get(url) + queryset = response.context["adminform"].form.fields["shelf"].queryset + self.assertEqual( + set(queryset.values_list("pk", flat=True)), expected_shelf_pks + ) + self._logout() + + def test_book_shelf_fk_autocomplete_view(self): data = self._create_multitenancy_test_env() self._test_multitenant_admin( - url=reverse("admin:testapp_book_add"), + url=self._get_autocomplete_view_path("testapp", "book", "shelf"), visible=[data["s1"].name], - hidden=[data["s2"].name, data["s3_inactive"].name], - select_widget=True, + hidden=[data["s2"].name], administrator=True, + # Keep disabled organizations hidden even for superusers. + superuser_hidden=[data["s3_inactive"].name], + ) + + def test_shelf_disabled_organization_admin_guard(self): + org = self._get_org() + shelf = self._create_shelf(name="disable-guard-shelf", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_admin_crud( + shelf, + change_data={"name": "renamed-shelf", "organization": str(org.pk)}, + roles=("superuser",), + ) + + def test_disabled_organization_mutating_action_is_blocked(self): + class ShelfActionAdmin(MultitenantAdminMixin, admin.ModelAdmin): + actions = ["rename_selected"] + + @admin.action(permissions=["change"]) + def rename_selected(self, request, queryset): + queryset.update(name="renamed-shelf") + + org = self._get_org() + shelf = self._create_shelf(name="action-guard-shelf", organization=org) + org.is_active = False + org.save() + request = RequestFactory().post( + "/", + {"action": "rename_selected", "_selected_action": [str(shelf.pk)]}, + ) + request.user = self._get_admin() + request._messages = CookieStorage(request) + model_admin = ShelfActionAdmin(Shelf, admin.site) + self.assertTrue(model_admin.has_delete_permission(request, shelf)) + model_admin.response_action(request, Shelf.objects.filter(pk=shelf.pk)) + shelf.refresh_from_db() + self.assertEqual(shelf.name, "action-guard-shelf") + + class ShelfActionExemptionAdmin(ShelfActionAdmin): + disabled_organization_action_exclusions = ("rename_selected",) + + request = RequestFactory().post( + "/", + {"action": "rename_selected", "_selected_action": [str(shelf.pk)]}, + ) + request.user = self._get_admin() + model_admin = ShelfActionExemptionAdmin(Shelf, admin.site) + model_admin.response_action(request, Shelf.objects.filter(pk=shelf.pk)) + shelf.refresh_from_db() + self.assertEqual(shelf.name, "renamed-shelf") + + def test_disabled_organization_guard_uses_custom_organization_resolver(self): + class LibraryActionAdmin(MultitenantAdminMixin, admin.ModelAdmin): + actions = ["rename_selected"] + + def get_object_organization(self, obj): + return obj.book.organization + + @admin.action(permissions=["change"]) + def rename_selected(self, request, queryset): + queryset.update(name="renamed-library") + + org = self._get_org() + book = self._create_book(name="resolver-book", organization=org) + library = self._create_library(name="resolver-library", book=book) + org.is_active = False + org.save() + request = RequestFactory().post( + "/", + {"action": "rename_selected", "_selected_action": [str(library.pk)]}, + ) + request.user = self._get_admin() + request._messages = CookieStorage(request) + model_admin = LibraryActionAdmin(Library, admin.site) + self.assertEqual(model_admin.has_change_permission(request, library), False) + model_admin.response_action(request, Library.objects.filter(pk=library.pk)) + library.refresh_from_db() + self.assertEqual(library.name, "resolver-library") + + def test_disabled_org_admin_crud_org_admin_loses_access(self): + org = self._create_org(name="admin-mixin-org-oa") + shelf = self._create_shelf(name="admin-mixin-shelf-oa", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_admin_crud( + shelf, + change_data={"name": "renamed", "organization": str(org.pk)}, + roles=("org_admin",), + ) + + def test_disabled_org_admin_crud_both_roles(self): + org = self._create_org(name="admin-mixin-org-both") + shelf = self._create_shelf(name="admin-mixin-shelf-both", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_admin_crud( + shelf, + change_data={"name": "renamed", "organization": str(org.pk)}, + create_data={"name": "new-shelf", "organization": str(org.pk)}, + ) + + def test_disabled_org_admin_crud_operations_subset(self): + org = self._create_org(name="admin-mixin-org-subset") + shelf = self._create_shelf(name="admin-mixin-shelf-subset", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_admin_crud( + shelf, + change_data={"name": "renamed", "organization": str(org.pk)}, + roles=("superuser",), + operations=("view",), + ) + self.assertEqual(self.shelf_model.objects.filter(pk=shelf.pk).exists(), True) + + def test_shelf_disabled_org_admin_inline_readonly(self): + data = self._create_multitenancy_test_env() + shelf_admin = ShelfAdmin(Shelf, admin.site) + self._test_disabled_org_admin_inline_readonly( + shelf_admin, data["s3_inactive"], active_obj=data["s1"] + ) + + def test_shelf_disabled_org_admin_inline_readonly_opt_out(self): + # The parent opt-out keeps BookInline writable. + data = self._create_multitenancy_test_env() + shelf_admin = ShelfDisabledOrgWriteAllowedAdmin(Shelf, admin.site) + request = RequestFactory().get("/") + request.user = self._get_admin() + + inlines = shelf_admin.get_inline_instances(request, data["s3_inactive"]) + for inline in inlines: + self.assertEqual( + inline.has_add_permission(request, data["s3_inactive"]), True + ) + self.assertEqual( + inline.has_change_permission(request, data["s3_inactive"]), True + ) + + def test_multitenant_parent_disabled_organization_guard(self): + data = self._create_multitenancy_test_env() + library_admin = LibraryParentAdmin(Library, admin.site) + request = RequestFactory().get("/") + request.user = self._get_admin() + active_library = Library.objects.create(name="lib-active", book=data["b1"]) + disabled_library = Library.objects.create( + name="lib-disabled", book=data["b3_inactive"] + ) + + with self.subTest("change allowed for object of active parent org"): + self.assertEqual( + library_admin.has_change_permission(request, active_library), True + ) + + with self.subTest("change blocked for object of disabled parent org"): + self.assertEqual( + library_admin.has_change_permission(request, disabled_library), False + ) + + with self.subTest("delete still allowed for object of disabled parent org"): + self.assertEqual( + library_admin.has_delete_permission(request, disabled_library), True + ) + + def test_multitenant_parent_disabled_organization_guard_http(self): + org = self._create_org(name="admin-mixin-org-parent") + book = self._create_book(name="parent-book", organization=org) + library = self._create_library(name="parent-library", book=book) + org.is_active = False + org.save() + self._test_disabled_org_admin_crud( + library, + change_data={ + "name": "renamed", + "address": "", + "book": str(book.pk), + }, + organization=org, + org_admin_expected={ + "view": {"status": 403}, + "change": {"status": 403, "unchanged": True}, + "delete": {"status": 403, "exists_after": True}, + }, + ) + + def test_add_permission_hidden_without_active_managed_org(self): + disabled_org = self._create_org(name="operator-disabled-org", is_active=False) + active_org = self._create_org(name="operator-active-org") + operator = self._create_operator() + operator.user_permissions.add(Permission.objects.get(codename="add_shelf")) + shelf_admin = ShelfAdmin(Shelf, admin.site) + request = RequestFactory().get("/") + + with self.subTest("no active managed org hides the Add button"): + self._create_org_user( + user=operator, organization=disabled_org, is_admin=True + ) + request.user = User.objects.get(pk=operator.pk) + self.assertEqual(shelf_admin.has_add_permission(request), False) + + with self.subTest("an active managed org restores the Add button"): + self._create_org_user(user=operator, organization=active_org, is_admin=True) + request.user = User.objects.get(pk=operator.pk) + self.assertEqual(shelf_admin.has_add_permission(request), True) + + def test_disabled_organization_write_protection_opt_out(self): + org = self._get_org() + shelf = self._create_shelf(name="opt-out-shelf", organization=org) + org.is_active = False + org.save() + shelf_admin = ShelfDisabledOrgWriteAllowedAdmin(Shelf, admin.site) + request = RequestFactory().get("/") + request.user = self._get_admin() + + with self.subTest("change permission is not blocked for the opted-out admin"): + self.assertEqual(shelf_admin.has_change_permission(request, shelf), True) + + with self.subTest("the disabled organization stays in the field's choices"): + form_class = shelf_admin.get_form(request, shelf) + org_field = form_class.base_fields["organization"] + self.assertIn(org.pk, org_field.queryset.values_list("pk", flat=True)) + + with self.subTest("the form can still be saved"): + form_class = shelf_admin.get_form(request, shelf) + form = form_class( + data={"name": shelf.name, "organization": org.pk}, instance=shelf + ) + self.assertTrue(form.is_valid(), form.errors) + form.save() + shelf.refresh_from_db() + self.assertEqual(shelf.organization_id, org.pk) + + def test_disabled_org_admin_crud_opt_out_override(self): + org = self._create_org(name="admin-mixin-org-optout") + config = self._create_config(name="optout-config", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_admin_crud( + config, + change_data={ + "name": "renamed-config", + "organization": str(org.pk), + }, + roles=("superuser",), + operations=("change",), + superuser_expected={"change": {"status": 200, "unchanged": False}}, + ) + config.refresh_from_db() + self.assertEqual(config.name, "renamed-config") + + def test_disabled_org_admin_org_field_excludes_disabled(self): + active_org = self._create_org(name="admin-mixin-active-org") + disabled_org = self._create_org( + name="admin-mixin-disabled-org", is_active=False + ) + add_url = reverse("admin:testapp_shelf_add") + self._test_disabled_org_admin_org_field_excludes_disabled( + add_url, + disabled_org, + roles=("superuser", "org_admin"), + organization=active_org, ) diff --git a/tests/testapp/tests/test_permission_classes.py b/tests/testapp/tests/test_permission_classes.py index 395700571..41a91e35a 100644 --- a/tests/testapp/tests/test_permission_classes.py +++ b/tests/testapp/tests/test_permission_classes.py @@ -1,12 +1,20 @@ +import json + from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission -from django.test import TestCase +from django.test import RequestFactory, TestCase from django.urls import reverse +from rest_framework.generics import ListAPIView +from rest_framework.test import APIRequestFactory, force_authenticate from swapper import load_model +from openwisp_users.api.mixins import FilterByOrganizationManaged, FilterByParentManaged +from openwisp_users.api.permissions import DisabledOrgReadOnly from openwisp_users.api.throttling import AuthRateThrottle -from ..models import Template +from ..models import Shelf, Template +from ..serializers import BookManagerSerializer +from ..views import TemplateDetailView, TemplateDisabledOrgWriteAllowedDetailView from .mixins import TestMultitenancyMixin User = get_user_model() @@ -363,3 +371,270 @@ def test_org_user_access_shared_object(self): "option": 200, }, ) + + def test_bare_protected_api_mixin_view_blocks_disabled_org_write(self): + org = self._get_org() + template = self._create_template(organization=org) + org.is_active = False + org.save() + admin = self._get_admin() + token = self._obtain_auth_token(username=admin) + auth = dict(HTTP_AUTHORIZATION=f"Bearer {token}") + detail_url = reverse("test_protected_template_detail", args=[template.pk]) + self._test_disabled_org_api_update( + detail_url, + auth, + {"name": "renamed"}, + template, + error_contains=str(DisabledOrgReadOnly.message), + ) + self._test_disabled_org_api_retrieve(detail_url, auth) + + def test_disabled_org_read_only_permission(self): + org = self._get_org() + template = self._create_template(organization=org) + org.is_active = False + org.save() + admin = self._get_admin() + token = self._obtain_auth_token(username=admin) + auth = dict(HTTP_AUTHORIZATION=f"Bearer {token}") + detail_url = reverse("test_template_detail", args=[template.pk]) + allowed_url = reverse( + "test_template_disabled_org_write_allowed_detail", args=[template.pk] + ) + self._test_disabled_org_api_update( + detail_url, + auth, + {"name": "renamed"}, + template, + error_contains=str(DisabledOrgReadOnly.message), + ) + self._test_disabled_org_api_retrieve(detail_url, auth) + + with self.subTest("opt-out view allows write"): + response = self.client.put( + allowed_url, + data={"name": "renamed", "organization": str(org.pk)}, + content_type="application/json", + **auth, + ) + self.assertEqual(response.status_code, 200) + + with self.subTest("manager update keeps the disabled organization"): + + class ManagerAllowedTemplateDetailView( + TemplateDisabledOrgWriteAllowedDetailView + ): + permission_classes = () + + def get_queryset(self): + return Template.objects.all() + + manager = self._create_administrator() + request = APIRequestFactory().put( + "/", + {"name": "renamed", "organization": str(org.pk)}, + format="json", + ) + force_authenticate(request, user=manager) + response = ManagerAllowedTemplateDetailView.as_view()( + request, pk=template.pk + ) + self.assertEqual(response.status_code, 200) + template.refresh_from_db() + self.assertEqual(template.organization_id, org.pk) + + with self.subTest("shared object unaffected"): + shared_template = self._create_template( + name="shared-template", organization=None + ) + shared_url = reverse("test_template_detail", args=[shared_template.pk]) + response = self.client.put( + shared_url, + data={"name": "shared-renamed"}, + content_type="application/json", + **auth, + ) + self.assertEqual(response.status_code, 200) + + with self.subTest("DELETE allowed"): + response = self.client.delete(detail_url, **auth) + self.assertEqual(response.status_code, 204) + + def test_disabled_org_api_crud_both_roles(self): + org = self._create_org(name="api-mixin-org-both") + template = self._create_template(name="t-both", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_api_crud( + template, + detail_url=reverse("test_template_detail", args=[template.pk]), + list_url=reverse("test_template_list"), + create_payload={"name": "t-both-new", "organization": str(org.pk)}, + update_payload={"name": "t-both-upd"}, + org_admin_expected={ + "list": {"status": 200, "object_present": False}, + "retrieve": {"status": 404}, + "create": { + "status": 400, + "error_field": "organization", + "error_contains": "does not exist or is disabled", + }, + "update": {"status": 404, "unchanged": True}, + "delete": {"status": 404, "exists_after": True}, + }, + ) + + def test_disabled_org_api_crud_session_auth(self): + org = self._create_org(name="api-mixin-org-session") + template = self._create_template(name="t-sess", organization=org) + org.is_active = False + org.save() + self._test_disabled_org_api_crud( + template, + detail_url=reverse("test_protected_template_detail", args=[template.pk]), + roles=("superuser",), + operations=("retrieve", "update"), + update_payload={"name": "t-sess-upd"}, + auth_mechanism="session", + ) + + def test_disabled_org_api_crud_opt_out_override(self): + org = self._create_org(name="api-mixin-org-optout") + template = self._create_template(name="t-optout", organization=org) + detail_url = reverse("test_template_detail", args=[template.pk]) + allowed_url = reverse( + "test_template_disabled_org_write_allowed_detail", args=[template.pk] + ) + org.is_active = False + org.save() + self._test_disabled_org_api_crud( + template, + detail_url=detail_url, + roles=("superuser",), + operations=("retrieve", "update"), + update_payload={"name": "renamed"}, + ) + self._test_disabled_org_api_crud( + template, + detail_url=allowed_url, + roles=("superuser",), + operations=("update",), + update_payload={"name": "t-optout-up"}, + superuser_expected={"update": {"status": 200, "unchanged": False}}, + ) + template.refresh_from_db() + self.assertEqual(template.name, "t-optout-up") + + admin = self._get_admin() + token = self._obtain_auth_token(username=admin) + auth = dict(HTTP_AUTHORIZATION=f"Bearer {token}") + with self.subTest("shared object unaffected"): + shared_template = self._create_template( + name="shared-template", organization=None + ) + shared_url = reverse("test_template_detail", args=[shared_template.pk]) + response = self.client.put( + shared_url, + data={"name": "shared-renamed"}, + content_type="application/json", + **auth, + ) + self.assertEqual(response.status_code, 200) + + with self.subTest("DELETE allowed"): + response = self.client.delete(detail_url, **auth) + self.assertEqual(response.status_code, 204) + + def test_organization_field_excludes_disabled_org(self): + disabled_org = self._create_org(name="disabled-org", is_active=False) + admin = self._get_admin() + token = self._obtain_auth_token(username=admin) + auth = dict(HTTP_AUTHORIZATION=f"Bearer {token}") + response = self.client.post( + reverse("test_template_list"), + data={"name": "t1", "organization": str(disabled_org.pk)}, + content_type="application/json", + **auth, + ) + self.assertEqual(response.status_code, 400) + self.assertIn( + "does not exist or is disabled", str(response.data["organization"][0]) + ) + self.assertEqual(self.template_model.objects.filter(name="t1").count(), 0) + + def test_fk_field_excludes_disabled_org_for_superuser(self): + active_org = self._create_org(name="active-fk-org") + disabled_org = self._create_org(name="disabled-fk-org", is_active=False) + admin = self._get_admin() + shelf_active = Shelf(name="shelf-active", organization=active_org) + shelf_active.full_clean() + shelf_active.save() + shelf_disabled = Shelf(name="shelf-disabled", organization=disabled_org) + shelf_disabled.full_clean() + shelf_disabled.save() + request = APIRequestFactory().get("/") + request.user = admin + serializer = BookManagerSerializer(context={"request": request}) + shelf_pks = set( + serializer.fields["shelf"].queryset.values_list("pk", flat=True) + ) + self.assertIn(shelf_active.pk, shelf_pks) + self.assertNotIn(shelf_disabled.pk, shelf_pks) + + def test_disabled_org_read_only_denies_on_misconfigured_field(self): + class BrokenOrgFieldTemplateDetailView(TemplateDetailView): + # A bad relation path must fail closed to avoid granting write access. + organization_field = "nonexistent_field" + + org = self._get_org() + template = self._create_template(organization=org, name="original-name") + admin = self._get_admin() + token = self._obtain_auth_token(username=admin) + request = RequestFactory().put( + "/", + data=json.dumps({"name": "renamed"}), + content_type="application/json", + HTTP_AUTHORIZATION=f"Bearer {token}", + ) + response = BrokenOrgFieldTemplateDetailView.as_view()(request, pk=template.pk) + response.render() + self.assertEqual(response.status_code, 403) + template.refresh_from_db() + self.assertEqual(template.name, "original-name") + + def test_disabled_org_read_only_select_related_valid_field(self): + org = self._get_org() + self._create_template(organization=org) + admin = self._get_admin() + request = APIRequestFactory().get("/") + request.user = admin + view = TemplateDetailView() + view.request = request + queryset = view.get_queryset() + self.assertIn("organization", queryset.query.select_related) + + def test_invalid_select_related_organization_paths_are_skipped(self): + class ManyToManyOrganizationView(FilterByOrganizationManaged, ListAPIView): + queryset = Shelf.objects.all() + organization_field = "tags" + + class ManyToManyParentView(FilterByParentManaged, ListAPIView): + queryset = Shelf.objects.all() + organization_field = "tags" + + def get_parent_queryset(self): + return Shelf.objects.all() + + org = self._get_org() + shelf = Shelf(name="m2m-lookup-shelf", organization=org) + shelf.full_clean() + shelf.save() + admin = self._get_admin() + request = APIRequestFactory().get("/") + request.user = admin + for view_class in (ManyToManyOrganizationView, ManyToManyParentView): + with self.subTest(view=view_class.__name__): + view = view_class() + view.request = request + self.assertEqual(view.get_queryset().count(), 1) diff --git a/tests/testapp/tests/test_selenium.py b/tests/testapp/tests/test_selenium.py index 4d99b71cd..2f3ffa003 100644 --- a/tests/testapp/tests/test_selenium.py +++ b/tests/testapp/tests/test_selenium.py @@ -1,44 +1,71 @@ +from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.contrib.staticfiles.testing import StaticLiveServerTestCase from django.db.models import Q +from django.test import tag from django.urls import reverse +from selenium.common.exceptions import TimeoutException from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.select import Select from selenium.webdriver.support.ui import WebDriverWait from swapper import load_model +from openwisp_users import admin as openwisp_users_admin from openwisp_utils.test_selenium_mixins import SeleniumTestMixin +from ..admin import BioInline +from ..models import Bio from .mixins import TestMultitenancyMixin Organization = load_model("openwisp_users", "Organization") +OrganizationUser = load_model("openwisp_users", "OrganizationUser") +User = get_user_model() +@tag("selenium_tests") class TestOrganizationAutocompleteField( SeleniumTestMixin, TestMultitenancyMixin, StaticLiveServerTestCase ): + @classmethod + def setUpClass(cls): + openwisp_users_admin.UserAdmin.inlines.append(BioInline) + super().setUpClass() + + @classmethod + def tearDownClass(cls): + openwisp_users_admin.UserAdmin.inlines.remove(BioInline) + super().tearDownClass() + def setUp(self): self.admin = self._create_admin( username=self.admin_username, password=self.admin_password ) + def logout(self, driver=None): + super().logout(driver) + driver = driver or self.web_driver + try: + WebDriverWait(driver, 5).until( + EC.url_to_be(f"{self.live_server_url}{reverse('admin:logout')}") + ) + except TimeoutException: + self.fail( + "Browser failed to logout the user: URL did not change to logout page" + ) + def _test_multitenant_autocomplete_org_field( self, username, password, path, visible, hidden ): self.login(username=username, password=password) self.open(path) - self.web_driver.find_element( - By.CSS_SELECTOR, "#select2-id_organization-container" - ).click() + self.find_element(By.CSS_SELECTOR, "#select2-id_organization-container").click() WebDriverWait(self.web_driver, 2).until( EC.invisibility_of_element_located( (By.CSS_SELECTOR, ".select2-results__option.loading-results") ) ) - options = self.web_driver.find_elements( - By.CSS_SELECTOR, ".select2-results__option" - ) + options = self.find_elements(By.CSS_SELECTOR, ".select2-results__option") for option in options: self.assertIn(option.text, visible) self.assertNotIn(option.text, hidden) @@ -47,12 +74,15 @@ def test_book_add_form_organization_field(self): path = reverse("admin:testapp_book_add") org1 = self._create_org(name="org1") org2 = self._create_org(name="org2") + disabled_org = self._create_org(name="disabled-org", is_active=False) administrator = self._create_administrator( organizations=[org1], username="tester", password="tester" ) administrator.user_permissions.add( *Permission.objects.filter( - Q(codename__contains="shelf") | Q(codename="view_organization") + Q(codename__contains="shelf") + | Q(codename="view_organization") + | Q(codename__contains="book") ).values_list("id", flat=True), ) @@ -61,8 +91,10 @@ def test_book_add_form_organization_field(self): path=path, username=self.admin_username, password=self.admin_password, - visible=Organization.objects.values_list("name", flat=True), - hidden=[], + visible=Organization.objects.exclude(id=disabled_org.id).values_list( + "name", flat=True + ), + hidden=[disabled_org.name], ) self.logout() @@ -76,9 +108,7 @@ def test_book_add_form_organization_field(self): "name", flat=True ), ) - org_select = Select( - self.web_driver.find_element(By.CSS_SELECTOR, "#id_organization") - ) + org_select = Select(self.find_element(By.CSS_SELECTOR, "#id_organization")) self.assertEqual(len(org_select.all_selected_options), 1) self.assertEqual(org_select.first_selected_option.text, org1.name) self.logout() @@ -95,9 +125,7 @@ def test_book_add_form_organization_field(self): id__in=[org1.id, org2.id] ).values_list("name", flat=True), ) - org_select = Select( - self.web_driver.find_element(By.CSS_SELECTOR, "#id_organization") - ) + org_select = Select(self.find_element(By.CSS_SELECTOR, "#id_organization")) self.assertEqual(len(org_select.all_selected_options), 0) self.logout() @@ -137,9 +165,146 @@ def test_shelf_add_form_organization_field(self): ) + ["Shared systemwide (no organization)"], ) - org_select = Select( - self.web_driver.find_element(By.CSS_SELECTOR, "#id_organization") - ) + org_select = Select(self.find_element(By.CSS_SELECTOR, "#id_organization")) self.assertEqual(len(org_select.all_selected_options), 1) self.assertEqual(org_select.first_selected_option.text, org1.name) self.logout() + + def test_user_add_form_does_not_hang(self): + path = reverse(f"admin:{User._meta.app_label}_user_add") + self.login(username=self.admin_username, password=self.admin_password) + self.open(path) + WebDriverWait(self.web_driver, 5).until( + EC.presence_of_element_located( + (By.CSS_SELECTOR, "select[id$='-organization'] + span.select2") + ) + ) + self.logout() + + def _create_disabled_bio(self, username): + organization = self._create_org(name=f"disabled-{username}-org") + user = self._create_user(username=username, email=f"{username}@example.com") + bio = Bio.objects.create( + user=user, organization=organization, website="https://example.com" + ) + organization.is_active = False + organization.save() + inline_prefix = Bio._meta.get_field("user").remote_field.get_accessor_name() + path = reverse(f"admin:{User._meta.app_label}_user_change", args=[user.pk]) + return bio, inline_prefix, path, user, organization + + def test_user_admin_disabled_org_bio(self): + with self.subTest("saving user fields"): + bio, inline_prefix, path, user, organization = self._create_disabled_bio( + "disabled-bio-save" + ) + self.login(username=self.admin_username, password=self.admin_password) + self.open(path) + organization_field = self.find_element( + By.ID, f"id_{inline_prefix}-0-organization" + ) + self.assertEqual( + organization_field.get_attribute("value"), str(organization.pk) + ) + self.assertEqual(organization_field.get_attribute("disabled"), "true") + self.assertEqual( + self.find_element(By.ID, f"id_{inline_prefix}-0-website").get_attribute( + "disabled" + ), + "true", + ) + notes_field = self.find_element(By.ID, "id_notes") + notes_field.send_keys("Updated notes") + save_button = self.find_element(By.NAME, "_continue") + self.web_driver.execute_script( + "arguments[0].scrollIntoView({block: 'center'});", save_button + ) + save_button.click() + self.find_element( + By.ID, f"id_{inline_prefix}-0-DELETE", timeout=10, wait_for="presence" + ) + user.refresh_from_db() + self.assertEqual(user.notes, "Updated notes") + self.assertEqual(Bio.objects.filter(pk=bio.pk).count(), 1) + self.assertEqual(bio.organization_id, organization.pk) + self.logout() + + with self.subTest("deleting the disabled-organization bio"): + bio, inline_prefix, path, user, organization = self._create_disabled_bio( + "disabled-bio-delete" + ) + self.login(username=self.admin_username, password=self.admin_password) + self.open(path) + delete_field = self.find_element( + By.ID, f"id_{inline_prefix}-0-DELETE", timeout=10, wait_for="presence" + ) + self.assertEqual(delete_field.is_enabled(), True) + self.find_element( + By.CSS_SELECTOR, f"label[for='id_{inline_prefix}-0-DELETE']" + ).click() + self.assertEqual(delete_field.is_selected(), True) + save_button = self.find_element(By.NAME, "_save") + self.web_driver.execute_script( + "arguments[0].scrollIntoView({block: 'center'});", save_button + ) + save_button.click() + WebDriverWait(self.web_driver, 5).until( + EC.presence_of_element_located( + (By.CSS_SELECTOR, ".messagelist .success") + ) + ) + self.assertEqual(Bio.objects.filter(pk=bio.pk).count(), 0) + user.refresh_from_db() + self.assertEqual(user.username, "disabled-bio-delete") + self.assertEqual(organization.is_active, False) + self.logout() + + def test_dynamic_organization_inline_normalizes_shared_value_on_submit(self): + # OrganizationUser.organization is a required field, so selecting + # "Shared systemwide (no organization)" on an inline row must be + # rejected with a validation error rather than saved as None. + path = reverse(f"admin:{User._meta.app_label}_user_add") + username = "shared-inline-user" + app_label = OrganizationUser._meta.app_label + organization = self._create_org(name="inline-organization") + self.login(username=self.admin_username, password=self.admin_password) + self.open(path) + self.find_element( + By.CSS_SELECTOR, f"#{app_label}_organizationuser-group .add-row a" + ).click() + static_org_field = self.find_element( + By.ID, f"id_{app_label}_organizationuser-0-organization" + ) + dynamic_org_field = WebDriverWait(self.web_driver, 5).until( + EC.presence_of_element_located( + (By.ID, f"id_{app_label}_organizationuser-1-organization") + ) + ) + self.web_driver.execute_script( + "var organization = new Option(arguments[1], arguments[2], true, true); " + "django.jQuery(arguments[0]).append(organization).trigger('change');", + static_org_field, + organization.name, + str(organization.pk), + ) + self.web_driver.execute_script( + "var shared = new Option(" + "'Shared systemwide (no organization)', 'null', true, true); " + "django.jQuery(arguments[0]).append(shared).trigger('change');", + dynamic_org_field, + ) + self.find_element(By.ID, "id_username").send_keys(username) + self.find_element(By.ID, "id_email").send_keys("test@openwisp.org") + self.find_element(By.ID, "id_password1").send_keys("testpassword") + self.find_element(By.ID, "id_password2").send_keys("testpassword") + self.find_element(By.CSS_SELECTOR, "input[name='_save']").click() + error = WebDriverWait(self.web_driver, 5).until( + EC.presence_of_element_located( + (By.CSS_SELECTOR, f"#{app_label}_organizationuser-1 .errorlist") + ) + ) + self.assertIn("This field is required", error.text) + self.assertFalse( + OrganizationUser.objects.filter(user__username=username).exists() + ) + self.logout() diff --git a/tests/testapp/tests/test_views.py b/tests/testapp/tests/test_views.py index f72e1e038..87820d9cc 100644 --- a/tests/testapp/tests/test_views.py +++ b/tests/testapp/tests/test_views.py @@ -64,6 +64,32 @@ def test_autocomplete_view_blank_option(self): else: self.fail("Null option not found in response") + def test_autocomplete_view_excludes_disabled_organization(self): + org1 = self._create_org(name="org1") + org2 = self._create_org(name="org2", is_active=False) + admin = self._get_admin() + self.client.force_login(admin) + path = self._get_autocomplete_view_path("testapp", "book", "organization") + + with self.subTest("widget path excludes disabled org"): + response = self.client.get(path + "&exclude_disabled=true") + ids = [option["id"] for option in response.json()["results"]] + self.assertIn(str(org1.pk), ids) + self.assertNotIn(str(org2.pk), ids) + + with self.subTest("list filter path keeps disabled org"): + response = self.client.get(path) + ids = [option["id"] for option in response.json()["results"]] + self.assertIn(str(org1.pk), ids) + self.assertIn(str(org2.pk), ids) + + with self.subTest("exclude_disabled=false keeps disabled org"): + # Treat only "true" as opt-in; "false" must leave results unchanged. + response = self.client.get(path + "&exclude_disabled=false") + ids = [option["id"] for option in response.json()["results"]] + self.assertIn(str(org1.pk), ids) + self.assertIn(str(org2.pk), ids) + def test_autocomplete_view_for_inline_admin(self): admin = self._get_admin() self.client.force_login(admin) diff --git a/tests/testapp/urls.py b/tests/testapp/urls.py index bf285fc39..c7808b203 100644 --- a/tests/testapp/urls.py +++ b/tests/testapp/urls.py @@ -59,6 +59,16 @@ views.template_detail, name="test_template_detail", ), + path( + "template_disabled_org_write_allowed//", + views.template_disabled_org_write_allowed_detail, + name="test_template_disabled_org_write_allowed_detail", + ), + path( + "protected_template//", + views.protected_template_detail, + name="test_protected_template_detail", + ), path( "library/", views.library_list, diff --git a/tests/testapp/views.py b/tests/testapp/views.py index 77d2a7994..0429a4637 100644 --- a/tests/testapp/views.py +++ b/tests/testapp/views.py @@ -22,9 +22,11 @@ FilterByParentMembership, FilterByParentOwned, FilterDjangoByOrgManaged, + ProtectedAPIMixin, ) from openwisp_users.api.permissions import ( BaseOrganizationPermission, + DisabledOrgReadOnly, DjangoModelPermissions, IsOrganizationManager, IsOrganizationMember, @@ -211,6 +213,7 @@ class TemplateListCreateView(FilterByOrganizationManaged, ListCreateAPIView): permission_classes = ( IsOrganizationMember, DjangoModelPermissions, + DisabledOrgReadOnly, ) queryset = Template.objects.all() @@ -221,10 +224,24 @@ class TemplateDetailView(FilterByOrganizationManaged, RetrieveUpdateDestroyAPIVi permission_classes = ( IsOrganizationMember, DjangoModelPermissions, + DisabledOrgReadOnly, ) queryset = Template.objects.all() +class TemplateDisabledOrgWriteAllowedDetailView(TemplateDetailView): + allow_disabled_organization_writes = True + + +class ProtectedTemplateDetailView( + ProtectedAPIMixin, FilterByOrganizationManaged, RetrieveUpdateDestroyAPIView +): + """Use the mixin defaults to cover inherited disabled-org protection.""" + + serializer_class = TemplateSerializer + queryset = Template.objects.all() + + class LibraryListFilter(FilterDjangoByOrgManaged): class Meta: model = Library @@ -285,6 +302,10 @@ class ShelfWithReadOnlyOrgListCreateView( shelf_list_owner_view = ShelfListOwnerView.as_view() template_list = TemplateListCreateView.as_view() template_detail = TemplateDetailView.as_view() +template_disabled_org_write_allowed_detail = ( + TemplateDisabledOrgWriteAllowedDetailView.as_view() +) +protected_template_detail = ProtectedTemplateDetailView.as_view() library_list = LibraryListCreateView.as_view() library_detail = LibraryDetailView.as_view() book_nested_shelf = BookNestedShelfListCreateView.as_view()