Coverage for sm/mixins.py: 92%
226 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-24 19:06 +0000
1from django.db.models import Q
2from django.contrib import messages
3from django.utils.translation import gettext as _
4from typing import Any, Optional
5from django.db.models.query import QuerySet
6from django.forms import ModelForm
7from django.contrib.auth.models import Group
8from django.db import transaction
9from django.core.exceptions import PermissionDenied
10from django.views.generic.edit import CreateView, UpdateView, DeleteView
11from rest_framework.exceptions import PermissionDenied as DRFPermissionDenied
12from rest_framework.exceptions import ValidationError
15def required_permission_codename(view: Any) -> str:
16 """
17 Returns the ``add``/``change``/``delete`` permission suffix a view type
18 requires (``"view"`` for everything else, e.g. list/detail/search).
19 """
20 if isinstance(view, DeleteView):
21 return "delete"
22 if isinstance(view, UpdateView):
23 return "change"
24 if isinstance(view, CreateView):
25 return "add"
26 return "view"
29def get_accessible_group_ids(request: Any) -> Optional[list[int]]:
30 """
31 Returns the group IDs a user may query, based on their session-selected
32 groups -- but ONLY for groups the user is actually a member of.
34 Selected groups that don't belong to the user are dropped so a user can
35 never read another tenant's data by tampering with the session.
37 Returns None if no selection is active (callers then fall back to the
38 user's own groups). If the user's active selection contains no group they
39 belong to (e.g. a tampered session), it falls back to None as well so the
40 caller shows only the user's own groups -- never another tenant's data.
41 """
42 user_groups = request.user.groups.all()
43 user_group_ids = set(user_groups.values_list("id", flat=True))
45 selected = request.session.get("selected_groups", [])
46 if not selected:
47 return None
49 accessible = [int(g) for g in selected if g.isdigit() and int(g) in user_group_ids]
50 # A selection that sanitizes down to nothing means the user only picked
51 # groups they don't belong to; fall back to their own groups.
52 return accessible or None
55def filter_queryset_by_tenant(queryset: QuerySet, request: Any) -> QuerySet:
56 """
57 Filters a queryset so the requesting user only sees data belonging to
58 their groups (or global items with no group). Superusers see everything.
59 Session-selected groups are honored but sanitized to groups the user
60 actually belongs to, so a user can never read another tenant's data.
61 """
62 if request.user.is_superuser:
63 return queryset
65 user_groups = request.user.groups.all()
66 accessible_group_ids = get_accessible_group_ids(request)
68 if accessible_group_ids is not None:
69 return queryset.filter(
70 Q(group__id__in=accessible_group_ids) | Q(group__isnull=True)
71 )
73 return queryset.filter(Q(group__in=user_groups) | Q(group__isnull=True))
76def filter_queryset_for_user(queryset: QuerySet, user: Any) -> QuerySet:
77 """
78 Filters a queryset to only items the user may reference (their groups plus
79 global/group-less rows). Unlike ``filter_queryset_by_tenant`` this takes a
80 plain user (no request/session) so it can be used when building forms.
81 """
82 if user is not None and getattr(user, "is_superuser", False):
83 return queryset
84 user_groups = user.groups.all() if user is not None else Group.objects.none()
85 return queryset.filter(Q(group__in=user_groups) | Q(group__isnull=True))
88def filter_history_queryset_by_tenant(history_qs: QuerySet, request: Any) -> QuerySet:
89 """
90 Filters a simple-history queryset by the requesting user's groups (plus
91 global/group-less entries). Superusers see everything. Uses the same
92 sanitized group selection as ``filter_queryset_by_tenant``.
93 """
94 if request.user.is_superuser:
95 return history_qs
97 user_groups = request.user.groups.all()
98 accessible_group_ids = get_accessible_group_ids(request)
99 group_ids = (
100 accessible_group_ids
101 if accessible_group_ids is not None
102 else list(user_groups.values_list("id", flat=True))
103 )
105 if not group_ids:
106 return history_qs.none()
108 return history_qs.filter(Q(group_id__in=group_ids) | Q(group_id__isnull=True))
111class BulkActionMixin:
112 """
113 Adds a generic bulk-action form and the delete permission flag to the
114 context of list views. Templates can then render the shared bulk toolbar.
115 """
117 def get_context_data(self, **kwargs: Any) -> Any:
118 from .forms import BulkActionForm
120 context = super().get_context_data(**kwargs) # type: ignore
121 model = getattr(self, "model", None)
122 if model and hasattr(self, "object_list"):
123 context["bulk_form"] = BulkActionForm()
124 context["perms_delete_model"] = self.request.user.has_perm(
125 "{}.delete_{}".format(model._meta.app_label, model._meta.model_name)
126 )
127 return context
130def get_tenant_model_counts(group: Optional[Group]) -> int:
131 """Helper function to count tenant items across all models for quota checking."""
132 if not group:
133 return 0
135 from server.models import Model as Server
136 from cluster.models import Model as Cluster
137 from domain.models import Model as Domain
138 from vendor.models import Model as Vendor
139 from operatingsystem.models import Model as OS
140 from status.models import Model as Status
141 from location.models import Model as Location
142 from patchtime.models import Model as Patchtime
143 from servermodel.models import Model as ServerModel
144 from clusterpackage.models import Model as ClusterPackage
145 from clustersoftware.models import Model as ClusterSoftware
146 from clusterpackagetype.models import Model as ClusterPackageType
148 return (
149 Server.objects.filter(group=group).count()
150 + Cluster.objects.filter(group=group).count()
151 + Domain.objects.filter(group=group).count()
152 + Vendor.objects.filter(group=group).count()
153 + OS.objects.filter(group=group).count()
154 + Status.objects.filter(group=group).count()
155 + Location.objects.filter(group=group).count()
156 + Patchtime.objects.filter(group=group).count()
157 + ServerModel.objects.filter(group=group).count()
158 + ClusterPackage.objects.filter(group=group).count()
159 + ClusterSoftware.objects.filter(group=group).count()
160 + ClusterPackageType.objects.filter(group=group).count()
161 )
164class MultiTenantMixin:
165 """
166 Mixin to filter querysets by user groups and auto-assign group on save.
167 Enforces item quotas per group.
168 """
170 def _tenant_permission_model(self) -> Any:
171 """
172 The model used for permission checks and history. Views whose queryset
173 model differs from the app they belong to (e.g. grouped list views that
174 return Vendor rows) may set ``permission_model`` to keep permission
175 checks anchored to their own app.
176 """
177 return getattr(self, "permission_model", None) or getattr(self, "model", None)
179 # Set on subclasses to block duplicate items within the same scope
180 # (group, or the single global scope when group is null).
181 unique_per_group_fields: tuple = ()
183 def get_queryset(self) -> QuerySet:
184 # Check basic view permission for the model
185 model = self._tenant_permission_model()
186 if model and not self.request.user.is_superuser:
187 opts = model._meta
188 codename = f"view_{opts.model_name.lower()}"
189 if not self.request.user.has_perm(f"{opts.app_label}.{codename}"):
190 from django.core.exceptions import PermissionDenied
192 raise PermissionDenied
194 queryset = super().get_queryset() # type: ignore
195 return filter_queryset_by_tenant(queryset, self.request)
197 def dispatch(self, request, *args, **kwargs):
198 """
199 Server-side permission enforcement. Read requests require ``view``;
200 write requests (POST) require ``add``/``change``/``delete`` for the
201 relevant operation. Without this, template-only gating lets any
202 view-only user POST directly to create/update/delete endpoints.
203 """
204 model = self._tenant_permission_model()
205 if model and not request.user.is_superuser:
206 opts = model._meta
207 model_name = opts.model_name.lower()
208 if request.method == "POST":
209 action = required_permission_codename(self)
210 perm = f"{opts.app_label}.{action}_{model_name}"
211 else:
212 perm = f"{opts.app_label}.view_{model_name}"
213 if not request.user.has_perm(perm):
214 raise PermissionDenied
215 return super().dispatch(request, *args, **kwargs)
217 def get_context_data(self, **kwargs: Any) -> Any:
218 context = super().get_context_data(**kwargs) # type: ignore
219 model = self._tenant_permission_model()
220 # Only add history to context for ListViews that have it configured
221 if model and hasattr(model, "history") and hasattr(self, "object_list"):
222 history_qs = model.history.all()
223 history_qs = filter_history_queryset_by_tenant(history_qs, self.request)
224 context["recent_history"] = history_qs.order_by("-history_date")[:10]
225 return context
227 def get_form_kwargs(self):
228 kwargs = super().get_form_kwargs()
229 # Only Create/Update views use SMForm-based forms that can scope FK
230 # choices to a user; DeleteView's plain confirm form does not accept
231 # a ``user`` kwarg.
232 if required_permission_codename(self) in ("add", "change"):
233 kwargs["user"] = self.request.user
234 return kwargs
236 def get_form(self, form_class=None):
237 """
238 Pre-assign the user's group to a new (unbound) form instance so the
239 form's ``clean()`` can enforce per-group uniqueness before
240 ``form_valid`` runs. This avoids SuccessMessageMixin emitting a false
241 success message when the form is rejected.
242 """
243 form = super().get_form(form_class=form_class) # type: ignore
244 if not hasattr(form, "instance"):
245 return form
246 if (
247 not form.instance.pk
248 and not form.instance.group
249 and not self.request.user.is_superuser
250 ):
251 user_groups = self.request.user.groups.all()
252 if user_groups.exists():
253 form.instance.group = user_groups.first()
254 return form
256 def check_quota(self, group: Optional[Group]) -> bool:
257 if not group or not hasattr(group, "profile"):
258 return True
260 profile = group.profile
261 max_items = profile.max_items
263 # Count items across models with transaction to prevent race conditions
264 with transaction.atomic():
265 # Lock the group profile to prevent concurrent modifications
266 GroupProfile = group.profile.__class__
267 GroupProfile.objects.select_for_update().get(pk=profile.pk)
269 count = get_tenant_model_counts(group)
271 return count < max_items
273 def form_valid(self, form: ModelForm) -> Any:
274 # Delete confirm forms are plain ``Form``; resolve the target object
275 # from the view for those, falling back to the bound model instance.
276 instance = getattr(form, "instance", None)
277 if instance is None:
278 instance = getattr(self, "object", None)
279 is_new = instance is None or not instance.pk
281 # Global (group-less) seed fixtures are read-only reference data.
282 # Tenant users must never be able to modify or delete them.
283 if not is_new and not self.request.user.is_superuser:
284 if getattr(instance, "group", None) is None:
285 raise PermissionDenied
286 # The group is immutable on update (editable=False); the form never
287 # carries it, so nothing needs to be preserved here.
288 return super().form_valid(form) # type: ignore
290 # New item: auto-assign the first group if not set and not superuser
291 if is_new and not self.request.user.is_superuser:
292 if instance is not None and not instance.group:
293 user_groups = self.request.user.groups.all()
294 if user_groups.exists():
295 instance.group = user_groups.first()
296 # Users without a group may not create global rows
297 if instance is None or getattr(instance, "group", None) is None:
298 raise PermissionDenied
300 # Check quota for NEW items
301 if is_new:
302 if not self.check_quota(getattr(instance, "group", None)):
303 quota_limit = 0
304 if instance and hasattr(instance, "group") and instance.group:
305 quota_limit = instance.group.profile.max_items
306 messages.error(
307 self.request,
308 _("Quota exceeded for this group (%d items).") % quota_limit,
309 )
310 return self.form_invalid(form)
312 # Call super().form_valid(form) to let other mixins (like SuccessMessageMixin)
313 # or the base view handle the actual saving and response.
314 return super().form_valid(form) # type: ignore
317class APIMultiTenantMixin:
318 """
319 Mixin for DRF ViewSets to filter by user groups and auto-assign on create.
320 """
322 def get_queryset(self) -> QuerySet:
323 # Check basic view permission for the model
324 model = getattr(self, "model", None)
325 if model and not self.request.user.is_superuser:
326 opts = model._meta
327 codename = f"view_{opts.model_name.lower()}"
328 if not self.request.user.has_perm(f"{opts.app_label}.{codename}"):
329 from django.core.exceptions import PermissionDenied
331 raise PermissionDenied
333 queryset = super().get_queryset() # type: ignore
334 return filter_queryset_by_tenant(queryset, self.request)
336 def check_duplicate_per_group(
337 self, serializer: Any, group: Optional[Group]
338 ) -> None:
339 """
340 Rejects a create whose values collide with an existing item in the
341 same scope. The scope is the target group, or the single global scope
342 when the group is null. Because the DB UniqueConstraint(name, group)
343 treats NULL rows as always-distinct, this guard is required to prevent
344 duplicate *global* items (matching the web forms' behavior).
346 Fields are derived from the model's ``UniqueConstraint`` declarations
347 that include the ``group`` column, so it stays in sync with the DB.
348 """
349 model = getattr(serializer, "Meta", None)
350 model = getattr(model, "model", None)
351 if model is None:
352 return
353 from django.db.models import UniqueConstraint
355 for constraint in model._meta.constraints:
356 if not isinstance(constraint, UniqueConstraint):
357 continue
358 fields = list(constraint.fields)
359 if "group" not in fields:
360 continue
361 fields.remove("group")
362 if not fields:
363 continue
365 lookup = {}
366 missing = False
367 for field in fields:
368 value = serializer.validated_data.get(field)
369 # Nested related fields (e.g. SlugRelatedField) may be
370 # serialized as the object itself or the pk.
371 if hasattr(value, "pk"):
372 value = value.pk
373 if value is None or value == "":
374 missing = True
375 break
376 lookup[field] = value
377 if missing:
378 continue
380 qs = model.objects.filter(**lookup)
381 if group and group.pk:
382 qs = qs.filter(group=group)
383 else:
384 qs = qs.filter(group__isnull=True)
386 if qs.exists():
387 from rest_framework.exceptions import ValidationError
389 raise ValidationError(
390 _("An item with these values already exists in this scope.")
391 )
393 def perform_create(self, serializer: Any) -> None:
394 user_groups = self.request.user.groups.all()
395 group = user_groups.first() if user_groups.exists() else None
397 if not self.request.user.is_superuser:
398 if group is None:
399 raise DRFPermissionDenied(
400 _("You must belong to a group to create items.")
401 )
403 # Simple quota check for API with transaction
404 if group and hasattr(group, "profile"):
405 # Use transaction to ensure atomic count
406 with transaction.atomic():
407 # Lock the group profile to prevent concurrent modifications
408 GroupProfile = group.profile.__class__
409 GroupProfile.objects.select_for_update().get(pk=group.profile.pk)
411 count = get_tenant_model_counts(group)
412 if count >= group.profile.max_items:
413 raise ValidationError(_("Quota exceeded for this group."))
415 # Block duplicates within the same scope (group or global).
416 self.check_duplicate_per_group(serializer, group)
418 serializer.save(group=group)
419 else:
420 serializer.save()
422 def perform_update(self, serializer: Any) -> None:
423 # Global (group-less) seed fixtures are read-only for tenant users
424 if (
425 not self.request.user.is_superuser
426 and getattr(serializer.instance, "group", None) is None
427 ):
428 raise DRFPermissionDenied(
429 _("Global reference data cannot be modified by tenant users.")
430 )
431 serializer.save()
433 def perform_destroy(self, instance: Any) -> None:
434 if (
435 not self.request.user.is_superuser
436 and getattr(instance, "group", None) is None
437 ):
438 raise DRFPermissionDenied(
439 _("Global reference data cannot be deleted by tenant users.")
440 )
441 instance.delete()