Coverage for backend/django/core/managers.py: 84%
353 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from dataclasses import dataclass
2from contextvars import ContextVar
3from django.db import models
4from authentication.user.models import User
5from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType
6from core.auxiliary.models.Flowsheet import Flowsheet
7from core.validation import (
8 cache_access_result,
9 get_current_flowsheet,
10 write_access_checks_are_bypassed,
11)
12from authentication.user.AccessTable import AccessTable
13from rest_framework.exceptions import PermissionDenied
14from rest_framework.exceptions import APIException
15from typing import TypeVar
16t = TypeVar('t', bound=models.Model, covariant=True)
19_validated_relation_bulk_update = ContextVar(
20 "validated_relation_bulk_update",
21 default=False,
22)
25@dataclass(frozen=True)
26class FlowsheetAccessState:
27 """
28 Normalized access result shared between the manager and view-layer helpers.
29 """
30 has_read_access: bool
31 has_write_access: bool
34def get_flowsheet_access(user: User, flowsheet_id: int) -> FlowsheetAccessState:
35 """
36 Compute read/write access for a user on a flowsheet and cache it in request context.
38 :param user: User object
39 :param flowsheet_id: Flowsheet ID to check access for
41 :return: read/write access state
42 """
43 flowsheet = (
44 Flowsheet.objects
45 .select_related("project")
46 .filter(id=flowsheet_id)
47 .only("project_id", "project__owner_id", "flowsheet_template_type")
48 .first()
49 )
50 if flowsheet is None:
51 return FlowsheetAccessState(False, False)
53 is_project_owner = bool(
54 flowsheet.project_id
55 and flowsheet.project is not None
56 and flowsheet.project.owner_id == user.id
57 )
59 direct_access = AccessTable.objects.filter(
60 user_id=user.id,
61 project_id=flowsheet.project_id
62 ).values("read_only").first()
63 user_has_direct_access = is_project_owner or direct_access is not None
64 is_read_only_share = False if is_project_owner else bool(direct_access["read_only"]) if direct_access else False
66 # Check if this is a public flowsheet template that the user can edit (only admins can edit public templates)
67 user_can_edit_template = (
68 user.is_staff
69 and not user_has_direct_access
70 and flowsheet.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate
71 )
73 has_read_access = user_has_direct_access or user_can_edit_template
74 has_write_access = user_can_edit_template or (user_has_direct_access and not is_read_only_share)
76 ctx = get_current_flowsheet() or {}
77 if str(ctx.get("flowsheet")) == str(flowsheet_id):
78 # Reuse the computed values for the rest of the request so queryset-level
79 # guards do not need to repeat the same project access lookup.
80 cache_access_result(has_read_access=has_read_access, has_write_access=has_write_access)
82 return FlowsheetAccessState(
83 has_read_access=has_read_access,
84 has_write_access=has_write_access,
85 )
88def has_flowsheet_read_access(user: User, flowsheet_id: int) -> bool:
89 """
90 Convenience helper for read access checks.
91 """
92 return get_flowsheet_access(user=user, flowsheet_id=flowsheet_id).has_read_access
95def has_flowsheet_write_access(user: User, flowsheet_id: int) -> bool:
96 """
97 Convenience helper for write access checks.
98 """
99 return get_flowsheet_access(user=user, flowsheet_id=flowsheet_id).has_write_access
102def project_access_filter(
103 user: User,
104 *,
105 project_path: str | None = "project",
106 write: bool = False,
107) -> models.Q:
108 """
109 Return a reusable ORM predicate for project read or write access.
111 ``project_path`` is the relation prefix from the queried model to
112 ``Project``. Use ``None`` when querying ``Project`` directly.
113 """
115 prefix = f"{project_path}__" if project_path else ""
116 owner_key = f"{prefix}owner"
117 access_user_key = f"{prefix}access_list__user"
118 access_read_only_key = f"{prefix}access_list__read_only"
120 if write:
121 return models.Q(**{owner_key: user}) | models.Q(
122 **{
123 access_user_key: user,
124 access_read_only_key: False,
125 }
126 )
127 return models.Q(**{owner_key: user}) | models.Q(**{access_user_key: user})
130def _get_or_compute_access_state(ctx: dict) -> FlowsheetAccessState | None:
131 """
132 Read access state from the active request context, computing it lazily when
133 the manager/queryset is the first layer to ask for it.
134 """
135 flowsheet = ctx.get("flowsheet")
136 user = ctx.get("user")
138 if flowsheet is None or user is None: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 return None
141 has_read_access = ctx.get("has_read_access")
142 has_write_access = ctx.get("has_write_access")
144 if has_read_access is None or has_write_access is None:
145 return get_flowsheet_access(user=user, flowsheet_id=flowsheet)
147 return FlowsheetAccessState(
148 has_read_access=has_read_access,
149 has_write_access=has_write_access,
150 )
153class StaleFlowsheetState(APIException):
154 """Conflict raised when a request attempts to write after a restore."""
156 status_code = 409
157 default_detail = "The flowsheet changed while this request was in progress. Refresh and try again."
158 default_code = "stale_flowsheet_state"
161def _require_write_access(ctx: dict) -> None:
162 """Require stable-flowsheet write access for a request-scoped mutation."""
164 if ctx.get("bypass_write_checks") or write_access_checks_are_bypassed():
165 return
166 access_state = _get_or_compute_access_state(ctx)
167 if access_state is not None and not access_state.has_write_access:
168 raise PermissionDenied("User does not have write access to this flowsheet.")
171def _require_captured_state_current(ctx: dict) -> None:
172 """Reject writes captured against a state replaced by concurrent restore."""
174 flowsheet_id = ctx.get("flowsheet")
175 flowsheet_state_id = ctx.get("flowsheet_state")
176 if flowsheet_id is None or flowsheet_state_id is None: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 return
179 is_current = Flowsheet.objects.filter(
180 pk=flowsheet_id,
181 current_state_id=flowsheet_state_id,
182 current_state__role="working",
183 ).exists()
184 if not is_current:
185 raise StaleFlowsheetState()
188class AccessControlQuerySet(models.QuerySet):
189 """
190 QuerySet-level write guard for mutating bulk operations that bypass serializer logic.
191 """
192 def _require_write_access(self):
193 ctx = get_current_flowsheet() or {}
194 flowsheet = ctx.get("flowsheet")
195 user = ctx.get("user")
197 # This is only the case when tests use objects directly without going through a view.
198 if flowsheet is None or user is None:
199 return
200 _require_write_access(ctx)
201 _require_captured_state_current(ctx)
203 def update(self, **kwargs):
204 self._require_write_access()
205 if "flowsheet_state" in kwargs or "flowsheet_state_id" in kwargs: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 raise ValueError("Use the unrestricted state manager to move aggregate rows.")
207 from core.state_validation import validate_state_owned_relation_update
209 # ``QuerySet.bulk_update`` implements its SQL through ``update`` with
210 # generated Case expressions. The concrete objects have already been
211 # checked before that internal call, so do not reject those expressions.
212 if not _validated_relation_bulk_update.get():
213 validate_state_owned_relation_update(self, kwargs)
214 return super().update(**kwargs)
216 def delete(self):
217 self._require_write_access()
218 # The pre-delete signal can trust this exact queryset. Cascaded rows
219 # use the same origin but are intentionally validated by their root
220 # mutation instead of issuing one current-state query per child.
221 self._state_mutation_validated = True
222 return super().delete()
224 def bulk_update(self, objs, fields, batch_size=None):
225 self._require_write_access()
226 if "flowsheet_state" in fields or "flowsheet_state_id" in fields: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 raise ValueError("Use the unrestricted state manager to move aggregate rows.")
228 from core.state_validation import validate_state_owned_relations_bulk
230 objs = tuple(objs)
231 validate_state_owned_relations_bulk(objs)
232 token = _validated_relation_bulk_update.set(True)
233 try:
234 return super().bulk_update(objs, fields, batch_size=batch_size)
235 finally:
236 _validated_relation_bulk_update.reset(token)
239class AccessControlManager(models.Manager.from_queryset(AccessControlQuerySet)):
240 """
241 Custom object manager to enforce access control based on the current flowsheet and user context.
242 This manager overrides the default queryset to filter objects based on the flowsheet.
244 By default, object modification methods (update, delete, etc.) will be scoped only to the
245 active flowsheet that the user has access to. Creation methods check explicitly if the user has
246 access to the flowsheet before allowing creation.
247 """
249 def _require_write_access(self, ctx: dict):
250 _require_write_access(ctx)
251 _require_captured_state_current(ctx)
253 def create(self, **kwargs):
254 ctx = get_current_flowsheet() or {}
256 flowsheet_state = ctx.get("flowsheet_state")
257 user = ctx.get("user")
259 # This is only the case when tests uses the objects directly without going through a view
260 if flowsheet_state is None or user is None:
261 target_state_id = kwargs.get("flowsheet_state_id")
262 target_state = kwargs.get("flowsheet_state")
263 if target_state_id is None and target_state is not None:
264 target_state_id = target_state.pk
265 if target_state_id is not None and not Flowsheet.objects.filter(
266 current_state_id=target_state_id,
267 current_state__role="working",
268 ).exists():
269 raise StaleFlowsheetState()
270 return super().create(**kwargs)
272 # Always pin the created object to the flowsheet from request context so
273 # callers cannot create rows under some other flowsheet id in the payload.
274 kwargs["flowsheet_state_id"] = flowsheet_state
275 self._require_write_access(ctx)
277 return super().create(**kwargs)
279 def bulk_create(self, objs, **kwargs):
280 # Materialize once because validation must inspect every object before
281 # Django consumes the iterable. This also preserves generator inputs.
282 objs = tuple(objs)
283 ctx = get_current_flowsheet() or {}
285 flowsheet_state = ctx.get("flowsheet_state")
286 user = ctx.get("user")
288 # This is only the case when tests uses the objects directly without going through a view
289 if flowsheet_state is None or user is None:
290 from core.state_validation import validate_state_owned_relations_bulk
292 target_state_ids = {obj.flowsheet_state_id for obj in objs}
293 valid_state_ids = set(
294 Flowsheet.objects.filter(
295 current_state_id__in=target_state_ids,
296 current_state__role="working",
297 ).values_list("current_state_id", flat=True)
298 )
299 if target_state_ids - valid_state_ids: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 raise StaleFlowsheetState()
301 validate_state_owned_relations_bulk(objs)
302 return super().bulk_create(objs, **kwargs)
304 # Force each bulk-created object onto the active flowsheet for the same
305 # reason as create(): request context, not payload, is the source of truth.
306 for obj in objs:
307 obj.flowsheet_state_id = flowsheet_state
309 self._require_write_access(ctx)
310 from core.state_validation import validate_state_owned_relations_bulk
312 validate_state_owned_relations_bulk(objs)
313 return super().bulk_create(objs, **kwargs)
315 def get_queryset(self):
316 """
317 Override the get_queryset method to filter objects based on the current flowsheet and user access.
318 """
319 unscoped_qs = super().get_queryset()
320 current_qs = unscoped_qs.filter(
321 flowsheet_state__role="working",
322 flowsheet_state__flowsheet__current_state_id=models.F("flowsheet_state_id"),
323 )
324 ctx = get_current_flowsheet() or {}
326 flowsheet_state = ctx.get("flowsheet_state")
327 user = ctx.get("user")
328 write_intent = bool(ctx.get("write_intent"))
330 # This is only the case when tests uses the objects directly without going through a view
331 if flowsheet_state is None or user is None:
332 return current_qs
334 access_state = _get_or_compute_access_state(ctx)
335 if access_state is None: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 return current_qs.none()
338 # flowsheet_context() validates historical state ownership and role.
339 # Context-free and ordinary request reads retain the implicit current
340 # working-state boundary.
341 qs = unscoped_qs if ctx.get("historical_read") else current_qs
343 filter_kwargs = {"flowsheet_state_id": flowsheet_state}
345 # For users with read access but no write access (read-only share),
346 # fail mutating viewset actions with an explicit 403.
347 if write_intent and access_state.has_read_access and not access_state.has_write_access:
348 raise PermissionDenied("This flowsheet is shared with read-only access.")
350 if access_state.has_read_access:
351 return qs.filter(**filter_kwargs)
352 else:
353 return qs.none()
356class IdentityAccessControlQuerySet(models.QuerySet):
357 """Write-guarded queryset for stable operational audit records."""
359 def _require_write_access(self):
360 ctx = get_current_flowsheet() or {}
361 if ctx.get("flowsheet") is None or ctx.get("user") is None:
362 return
363 _require_write_access(ctx)
364 _require_captured_state_current(ctx)
366 def update(self, **kwargs):
367 self._require_write_access()
368 ownership_fields = {
369 "flowsheet",
370 "flowsheet_id",
371 "flowsheet_state",
372 "flowsheet_state_id",
373 }
374 if ownership_fields & kwargs.keys(): 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true
375 raise ValueError("Operational ownership cannot be moved through a queryset update.")
376 from core.state_validation import validate_state_owned_relation_update
378 validate_state_owned_relation_update(self, kwargs)
379 return super().update(**kwargs)
381 def delete(self):
382 self._require_write_access()
383 self._state_mutation_validated = True
384 return super().delete()
386 def bulk_update(self, objs, fields, batch_size=None):
387 self._require_write_access()
388 ownership_fields = {
389 "flowsheet",
390 "flowsheet_id",
391 "flowsheet_state",
392 "flowsheet_state_id",
393 }
394 if ownership_fields & set(fields):
395 raise ValueError("Operational ownership cannot be moved through a bulk update.")
396 from core.state_validation import validate_state_owned_relations_bulk
398 validate_state_owned_relations_bulk(objs)
399 return super().bulk_update(objs, fields, batch_size=batch_size)
402class IdentityAccessControlManager(models.Manager.from_queryset(IdentityAccessControlQuerySet)):
403 """Manager for audit rows owned by stable flowsheet identity.
405 Models carrying optional state provenance are pinned to the captured working
406 state on request-scoped creation, while reads remain available after that
407 state is retired and deleted.
408 """
410 def create(self, **kwargs):
411 ctx = get_current_flowsheet() or {}
412 flowsheet_id = ctx.get("flowsheet")
413 user = ctx.get("user")
414 if flowsheet_id is None or user is None:
415 return super().create(**kwargs)
417 kwargs["flowsheet_id"] = flowsheet_id
418 if any(field.name == "flowsheet_state" for field in self.model._meta.fields):
419 kwargs["flowsheet_state_id"] = ctx.get("flowsheet_state")
420 _require_write_access(ctx)
421 _require_captured_state_current(ctx)
422 return super().create(**kwargs)
424 def bulk_create(self, objs, **kwargs):
425 ctx = get_current_flowsheet() or {}
426 flowsheet_id = ctx.get("flowsheet")
427 user = ctx.get("user")
428 if flowsheet_id is None or user is None:
429 from core.state_validation import validate_state_owned_relations_bulk
431 validate_state_owned_relations_bulk(objs)
432 return super().bulk_create(objs, **kwargs)
434 has_state = any(field.name == "flowsheet_state" for field in self.model._meta.fields)
435 for obj in objs:
436 obj.flowsheet_id = flowsheet_id
437 if has_state: 437 ↛ 435line 437 didn't jump to line 435 because the condition on line 437 was always true
438 obj.flowsheet_state_id = ctx.get("flowsheet_state")
439 _require_write_access(ctx)
440 _require_captured_state_current(ctx)
441 from core.state_validation import validate_state_owned_relations_bulk
443 validate_state_owned_relations_bulk(objs)
444 return super().bulk_create(objs, **kwargs)
446 def get_queryset(self):
447 qs = super().get_queryset()
448 ctx = get_current_flowsheet() or {}
449 flowsheet_id = ctx.get("flowsheet")
450 user = ctx.get("user")
451 if flowsheet_id is None or user is None:
452 return qs
454 access_state = _get_or_compute_access_state(ctx)
455 if access_state is None: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 return qs.none()
457 if ctx.get("write_intent") and access_state.has_read_access and not access_state.has_write_access: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true
458 raise PermissionDenied("This flowsheet is shared with read-only access.")
459 if access_state.has_read_access:
460 return qs.filter(flowsheet_id=flowsheet_id)
461 return qs.none()
464class AllFlowsheetStatesQuerySet(models.QuerySet):
465 """Expose historical rows while reserving writes for lifecycle services."""
467 def _require_lifecycle_write(self) -> None:
468 """Require the explicit internal bypass before crossing state boundaries."""
470 if not write_access_checks_are_bypassed():
471 raise PermissionDenied(
472 "Cross-state mutations are restricted to internal lifecycle services."
473 )
475 def create(self, **kwargs):
476 self._require_lifecycle_write()
477 return super().create(**kwargs)
479 def bulk_create(self, objs, **kwargs):
480 self._require_lifecycle_write()
481 return super().bulk_create(objs, **kwargs)
483 def update(self, **kwargs):
484 self._require_lifecycle_write()
485 return super().update(**kwargs)
487 def bulk_update(self, objs, fields, batch_size=None):
488 self._require_lifecycle_write()
489 return super().bulk_update(objs, fields, batch_size=batch_size)
491 def update_or_create(self, defaults=None, **kwargs):
492 self._require_lifecycle_write()
493 return super().update_or_create(defaults=defaults, **kwargs)
495 def delete(self):
496 self._require_lifecycle_write()
497 return super().delete()
500class AllFlowsheetStatesManager(
501 models.Manager.from_queryset(AllFlowsheetStatesQuerySet)
502):
503 """Read every state while requiring an explicit bypass for mutations."""
505 use_in_migrations = True
507class SoftDeleteManager(AccessControlManager):
508 def __init__(self):
509 super().__init__()
510 def get_queryset(self):
511 # filter out deleted objects
512 return super().get_queryset().filter(is_deleted=False)
514 def include_deleted(self):
515 # include deleted objects
516 return super().get_queryset()
519class ProjectAccessControlQuerySet(models.QuerySet):
520 """Guard project-owned bulk mutations with the captured access state."""
522 def _require_write_access(self):
523 ctx = get_current_flowsheet() or {}
524 if ctx.get("flowsheet") is None or ctx.get("user") is None: 524 ↛ 526line 524 didn't jump to line 526 because the condition on line 524 was always true
525 return
526 _require_write_access(ctx)
527 _require_captured_state_current(ctx)
529 def update(self, **kwargs):
530 self._require_write_access()
531 return super().update(**kwargs)
533 def delete(self):
534 self._require_write_access()
535 return super().delete()
537 def bulk_update(self, objs, fields, batch_size=None):
538 self._require_write_access()
539 return super().bulk_update(objs, fields, batch_size=batch_size)
542class ProjectAccessControlManager(
543 models.Manager.from_queryset(ProjectAccessControlQuerySet)
544):
545 """
546 Access-control manager for project-owned models addressed through flowsheet context.
547 """
549 def _context_project_id(self, ctx: dict) -> int | None:
550 flowsheet = ctx.get("flowsheet")
551 if flowsheet is None:
552 return None
553 return (
554 Flowsheet.objects
555 .filter(pk=flowsheet)
556 .values_list("project_id", flat=True)
557 .first()
558 )
560 def _require_write_access(self, ctx: dict):
561 if ctx.get("bypass_write_checks") or write_access_checks_are_bypassed(): 561 ↛ 562line 561 didn't jump to line 562 because the condition on line 561 was never true
562 return
564 access_state = _get_or_compute_access_state(ctx)
565 if access_state is None: 565 ↛ 566line 565 didn't jump to line 566 because the condition on line 565 was never true
566 return
567 if not access_state.has_write_access: 567 ↛ 568line 567 didn't jump to line 568 because the condition on line 567 was never true
568 raise PermissionDenied("User does not have write access to this flowsheet.")
569 _require_captured_state_current(ctx)
571 def create(self, **kwargs):
572 ctx = get_current_flowsheet() or {}
573 project_id = self._context_project_id(ctx)
574 user = ctx.get("user")
576 if project_id is None or user is None:
577 return super().create(**kwargs)
579 kwargs["project_id"] = project_id
580 self._require_write_access(ctx)
581 return super().create(**kwargs)
583 def bulk_create(self, objs, **kwargs):
584 ctx = get_current_flowsheet() or {}
585 project_id = self._context_project_id(ctx)
586 user = ctx.get("user")
588 if project_id is None or user is None:
589 return super().bulk_create(objs, **kwargs)
591 for obj in objs:
592 obj.project_id = project_id
594 self._require_write_access(ctx)
595 return super().bulk_create(objs, **kwargs)
597 def get_queryset(self):
598 qs = super().get_queryset()
599 ctx = get_current_flowsheet() or {}
601 project_id = self._context_project_id(ctx)
602 user = ctx.get("user")
603 write_intent = bool(ctx.get("write_intent"))
605 if project_id is None or user is None:
606 return qs
608 access_state = _get_or_compute_access_state(ctx)
609 if access_state is None: 609 ↛ 610line 609 didn't jump to line 610 because the condition on line 609 was never true
610 return qs.none()
612 if write_intent and access_state.has_read_access and not access_state.has_write_access:
613 raise PermissionDenied("This project is shared with read-only access.")
615 if access_state.has_read_access:
616 return qs.filter(project_id=project_id)
617 return qs.none()
620class UserAccessControlManager(models.Manager.from_queryset(AccessControlQuerySet)):
621 """Scope reusable, user-owned records to the active request user."""
623 def create(self, **kwargs):
624 ctx = get_current_flowsheet() or {}
625 user = ctx.get("user")
626 if user is not None:
627 kwargs["owner"] = user
628 return super().create(**kwargs)
630 def bulk_create(self, objs, **kwargs):
631 ctx = get_current_flowsheet() or {}
632 user = ctx.get("user")
633 if user is not None:
634 for obj in objs:
635 obj.owner = user
636 return super().bulk_create(objs, **kwargs)
638 def get_queryset(self):
639 queryset = super().get_queryset()
640 user = (get_current_flowsheet() or {}).get("user")
641 return queryset.filter(owner=user) if user is not None else queryset
645def include_soft_deleted[t](objects: models.Manager[t]) -> models.QuerySet[t]:
646 """
647 In some operations, including copying a flowsheet,
648 we need to include the soft deleted objects
649 Example usage:
650 include_soft_deleted(SimulationObject.objects).filter(flowsheet_id=1)
651 """
652 if objects is not None and isinstance(objects, SoftDeleteManager):
653 return objects.include_deleted().all()
654 else:
655 return objects.all()