228 lines
8.9 KiB
Python
228 lines
8.9 KiB
Python
from functools import wraps
|
||
from typing import Optional, Set, Iterable
|
||
|
||
from django.db.models import Q
|
||
from django.http import HttpResponseForbidden
|
||
from django.shortcuts import get_object_or_404
|
||
|
||
from common.models import UserConfig, UserThematics
|
||
from contracts.models import CompanyMember
|
||
from .models import Observation
|
||
|
||
|
||
class ObservationAccessContext:
|
||
"""
|
||
Contexte d'accès aux observations pour un utilisateur donné.
|
||
|
||
Le filtre sur les créateurs (utilisateurs de la même société) est stocké
|
||
sous forme de QuerySet non évalué (_creator_qs) afin d'être injecté comme
|
||
sous-requête SQL dans filter_queryset(), évitant ainsi un gros IN-list.
|
||
L'évaluation Python n'a lieu que si allows() est appelé (vérification objet
|
||
par objet), et le résultat est mis en cache.
|
||
"""
|
||
|
||
def __init__(self, user_config, thematic_ids, process_thematic_ids=None, creator_qs=None):
|
||
self.user_config = user_config
|
||
self.thematic_ids = thematic_ids
|
||
self.process_thematic_ids = process_thematic_ids if process_thematic_ids is not None else set()
|
||
self._creator_qs = creator_qs # QuerySet(user_id) ou None – jamais évalué ici
|
||
self._creator_ids_cache = None # évalué paresseusement dans allows()
|
||
|
||
@property
|
||
def allowed_creator_ids(self) -> Optional[Set[int]]:
|
||
"""Ensemble des user_id autorisés – évalué une seule fois (pour allows())."""
|
||
if self._creator_qs is None:
|
||
return None
|
||
if self._creator_ids_cache is None:
|
||
self._creator_ids_cache = set(
|
||
self._creator_qs.values_list('user_id', flat=True)
|
||
)
|
||
self._creator_ids_cache.add(self.user_config.user_id)
|
||
return self._creator_ids_cache
|
||
|
||
def can_process_observation(self, observation) -> bool:
|
||
"""Vérifie si l'utilisateur peut traiter (changer le statut de) cette observation."""
|
||
if self.user_config.has_role('admin') or self.user_config.has_role('top_manager'):
|
||
return True
|
||
# Un manager ou controller interne ayant les droits de traitement pour cette
|
||
# thématique peut traiter toutes les observations liées à cette thématique,
|
||
# sans être bloqué par les restrictions d'équipes liées aux symptômes.
|
||
if (
|
||
self.user_config.is_intern
|
||
and self.process_thematic_ids
|
||
and observation.thematic_id in self.process_thematic_ids
|
||
and any(self.user_config.has_role(r) for r in ('manager', 'controller'))
|
||
):
|
||
return True
|
||
# Règle prioritaire : si le symptôme a des équipes assignées,
|
||
# l'utilisateur doit en faire partie pour pouvoir traiter l'observation.
|
||
if observation.symptom_id:
|
||
from interventions.models import Symptom
|
||
has_symptom_teams = Symptom.objects.filter(
|
||
pk=observation.symptom_id, teams__isnull=False
|
||
).exists()
|
||
if has_symptom_teams:
|
||
user_in_symptom_team = CompanyMember.objects.filter(
|
||
user_id=self.user_config.user_id,
|
||
teams__symptoms__id=observation.symptom_id,
|
||
).exists()
|
||
if not user_in_symptom_team:
|
||
return False
|
||
if self.user_config.has_role('operator'):
|
||
return True
|
||
if self.process_thematic_ids and observation.thematic_id in self.process_thematic_ids:
|
||
return True
|
||
return False
|
||
|
||
def filter_queryset(self, queryset):
|
||
# Role admin and top_manager can access all observations
|
||
if self.user_config.has_role('admin') or self.user_config.has_role('top_manager'):
|
||
return queryset
|
||
|
||
if not self.thematic_ids:
|
||
return queryset.none()
|
||
|
||
qs = queryset.filter(thematic_id__in=self.thematic_ids)
|
||
if self._creator_qs is not None:
|
||
# Sous-requête DB : évite un gros IN-list et inclut l'utilisateur lui-même
|
||
qs = qs.filter(
|
||
Q(created_by_id=self.user_config.user_id)
|
||
| Q(created_by__in=self._creator_qs)
|
||
)
|
||
return qs
|
||
|
||
def allows(self, observation: Observation) -> bool:
|
||
# Role admin and top_manager can access all observations
|
||
if self.user_config.has_role('admin') or self.user_config.has_role('top_manager'):
|
||
return True
|
||
|
||
if not self.thematic_ids or observation.thematic_id not in self.thematic_ids:
|
||
return False
|
||
|
||
if (
|
||
self._creator_qs is not None
|
||
and observation.created_by_id not in self.allowed_creator_ids
|
||
):
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def _get_company_creator_subquery(user):
|
||
"""
|
||
Retourne un QuerySet(user_id) des membres de la même société que l'utilisateur.
|
||
Aucune requête SQL n'est exécutée ici – il s'agit d'une sous-requête différée
|
||
destinée à être injectée dans filter_queryset().
|
||
"""
|
||
return CompanyMember.objects.filter(
|
||
teams__company__in=CompanyMember.objects.filter(user=user).values('teams__company')
|
||
).values('user_id')
|
||
|
||
|
||
def get_observation_access_context(user, user_config=None) -> Optional[ObservationAccessContext]:
|
||
"""
|
||
Construit le contexte d'accès aux observations.
|
||
|
||
user_config peut être passé en paramètre si déjà chargé (évite une requête
|
||
redondante, notamment depuis home/views.py).
|
||
"""
|
||
if user_config is None:
|
||
try:
|
||
user_config = UserConfig.objects.get(user=user)
|
||
except UserConfig.DoesNotExist:
|
||
return None
|
||
|
||
# Une seule requête pour les deux flags de permission thématique
|
||
ut_rows = list(
|
||
UserThematics.objects.filter(user_config=user_config)
|
||
.values('thematic_id', 'can_view_interventions', 'can_process_observations')
|
||
)
|
||
thematic_ids: Set[int] = {r['thematic_id'] for r in ut_rows if r['can_view_interventions']}
|
||
process_thematic_ids: Set[int] = {r['thematic_id'] for r in ut_rows if r['can_process_observations']}
|
||
|
||
creator_qs = None
|
||
if not user_config.is_intern:
|
||
creator_qs = _get_company_creator_subquery(user)
|
||
|
||
return ObservationAccessContext(
|
||
user_config=user_config,
|
||
thematic_ids=thematic_ids,
|
||
process_thematic_ids=process_thematic_ids,
|
||
creator_qs=creator_qs,
|
||
)
|
||
|
||
|
||
def observation_access_required(view_func):
|
||
@wraps(view_func)
|
||
def _wrapped_view(request, *args, **kwargs):
|
||
access_context = getattr(request, 'observation_access', None)
|
||
if access_context is None:
|
||
access_context = get_observation_access_context(request.user)
|
||
if access_context is None:
|
||
return HttpResponseForbidden("You do not have access to this resource.")
|
||
request.observation_access = access_context
|
||
return view_func(request, *args, **kwargs)
|
||
|
||
return _wrapped_view
|
||
|
||
|
||
def observation_object_permission_required(select_related: Optional[Iterable[str]] = None):
|
||
select_related = select_related or ('thematic', 'category', 'created_by')
|
||
|
||
def decorator(view_func):
|
||
@wraps(view_func)
|
||
def _wrapped_view(request, observation_id, *args, **kwargs):
|
||
access_context = getattr(request, 'observation_access', None)
|
||
if access_context is None:
|
||
access_context = get_observation_access_context(request.user)
|
||
if access_context is None:
|
||
return HttpResponseForbidden("You do not have access to this observation.")
|
||
request.observation_access = access_context
|
||
|
||
queryset = Observation.objects.all()
|
||
if select_related:
|
||
queryset = queryset.select_related(*select_related)
|
||
|
||
observation = get_object_or_404(queryset, pk=observation_id)
|
||
|
||
if not access_context.allows(observation):
|
||
return HttpResponseForbidden("You do not have access to this observation.")
|
||
|
||
request.observation_obj = observation
|
||
return view_func(request, observation_id, *args, **kwargs)
|
||
|
||
return _wrapped_view
|
||
|
||
return decorator
|
||
|
||
|
||
# =============================================================================
|
||
# Guard functions for private file access
|
||
# =============================================================================
|
||
|
||
def guard_observation_photo(obj, user):
|
||
"""
|
||
Guard function for ObservationPhoto.photo field.
|
||
Checks if user has access to the observation.
|
||
"""
|
||
from .models import ObservationPhoto
|
||
if not isinstance(obj, ObservationPhoto):
|
||
return False
|
||
|
||
if not user.is_authenticated:
|
||
return False
|
||
|
||
access_context = get_observation_access_context(user)
|
||
if access_context is None:
|
||
return False
|
||
|
||
return access_context.allows(obj.observation)
|
||
|
||
|
||
__all__ = [
|
||
'ObservationAccessContext',
|
||
'get_observation_access_context',
|
||
'observation_access_required',
|
||
'observation_object_permission_required',
|
||
'guard_observation_photo',
|
||
]
|