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', ]