feat: implement contract-based asset visibility restrictions for external users

loko/assets/permissions.py

@@ -156,14 +156,124 @@ def _check_instance_permission(user, obj, permission_field, thematic_fallback_fn
     return False
 
 
+def get_asset_contract_ids(asset):
+    """
+    Retourne l'ensemble des IDs de contrats associés à une instance d'asset ou de localisation,
+    ou None si le modèle d'asset ne gère pas de contrats.
+    """
+    if asset is None:
+        return None
+
+    contract_ids = set()
+    has_contract_relation = False
+
+    # 1. Clés étrangères directes
+    if hasattr(asset, 'maintenance_contract_id') and asset.maintenance_contract_id:
+        contract_ids.add(asset.maintenance_contract_id)
+        has_contract_relation = True
+    elif hasattr(asset, 'maintenance_contract'):
+        has_contract_relation = True
+
+    if hasattr(asset, 'controller_maintenance_contract_id') and asset.controller_maintenance_contract_id:
+        contract_ids.add(asset.controller_maintenance_contract_id)
+        has_contract_relation = True
+    elif hasattr(asset, 'controller_maintenance_contract'):
+        has_contract_relation = True
+
+    if hasattr(asset, 'contract_id') and asset.contract_id:
+        contract_ids.add(asset.contract_id)
+        has_contract_relation = True
+    elif hasattr(asset, 'contract'):
+        has_contract_relation = True
+
+    # 2. Relations ManyToMany ou Reverse managers (contracts, etc.)
+    if hasattr(asset, 'contracts'):
+        has_contract_relation = True
+        try:
+            model = asset.contracts.model
+            if model.__name__ == 'Contract':
+                for cid in asset.contracts.values_list('pk', flat=True):
+                    contract_ids.add(cid)
+            elif hasattr(model, 'contract_id'):
+                qs = asset.contracts.all()
+                if hasattr(model, 'status'):
+                    qs = qs.filter(status='active')
+                for cid in qs.values_list('contract_id', flat=True):
+                    contract_ids.add(cid)
+        except Exception:
+            pass
+
+    for rel_name in [
+        'trafficlight_contracts', 'structure_contracts', 'its_locations_contracts',
+        'its_assets_contracts', 'controlcenters_contracts', 'controlcenters_assets_contracts',
+        'clean_location_contracts'
+    ]:
+        if hasattr(asset, rel_name):
+            has_contract_relation = True
+            try:
+                manager = getattr(asset, rel_name)
+                if hasattr(manager.model, 'status'):
+                    cids = manager.filter(status='active').values_list('contract_id', flat=True)
+                else:
+                    cids = manager.values_list('contract_id', flat=True)
+                contract_ids.update(cids)
+            except Exception:
+                pass
+
+    # 3. Assets enfants liés à un parent (intersection, location, structure, control_center, pole, etc.)
+    for parent_attr in ['intersection', 'location', 'structure', 'control_center', 'building', 'pole']:
+        if hasattr(asset, parent_attr):
+            parent = getattr(asset, parent_attr)
+            if parent:
+                parent_cids = get_asset_contract_ids(parent)
+                if parent_cids is not None:
+                    has_contract_relation = True
+                    contract_ids.update(parent_cids)
+
+    if not has_contract_relation:
+        return None
+    return contract_ids
+
+
 def can_view_asset(user, asset):
     """Check if user can view a specific asset."""
+    if not user or not user.is_authenticated:
+        return False
+
+    try:
+        user_config = UserConfig.objects.get(user=user)
+    except UserConfig.DoesNotExist:
+        user_config = None
+
+    if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
+        return True
+    if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')):
+        return True
+
     def _fallback():
         thematic = _get_asset_thematic(asset)
         if not thematic:
             return False
         can_view, _ = get_user_asset_thematic_access(user, thematic)
-        return can_view
+        if not can_view:
+            return False
+
+        # Vérifier la restriction par contrat pour les utilisateurs externes
+        is_external = (user_config is None) or (not user_config.is_intern)
+        must_limit = is_external and (not user_config or user_config.limit_assets_to_contracts)
+        if must_limit and user_config:
+            asset_cids = get_asset_contract_ids(asset)
+            if asset_cids is not None:
+                from common.models import UserContractAccess
+                accessible_cids = set(
+                    UserContractAccess.objects.filter(
+                        user_config=user_config,
+                        can_view_assets=True
+                    ).values_list('contract_id', flat=True)
+                )
+                if not (asset_cids & accessible_cids):
+                    return False
+        return True
 
     return _check_instance_permission(user, asset, 'can_view', _fallback)
 
@@ -276,27 +386,12 @@ def can_delete_location(user, location):
 
 
 def _get_asset_thematic(asset):
-    """Get the thematic for an asset."""
+    """Get the thematic for an asset instance or model class."""
+    if asset is None:
+        return None
+
     from common.models import Thematic
-    
-    # Try from category
-    if hasattr(asset, 'category') and asset.category:
-        return asset.category.thematic
 
-    # Try direct thematic attribute
-    if hasattr(asset, 'thematic') and asset.thematic:
-        return asset.thematic
-
-    # Try from parent building category
-    if hasattr(asset, 'building') and asset.building and hasattr(asset.building, 'category') and asset.building.category:
-        return asset.building.category.thematic
-    
-    # Try from model's category
-    if hasattr(asset, 'model') and asset.model and hasattr(asset.model, 'category') and asset.model.category:
-        return asset.model.category.thematic
-    
-    # Fallback: deduce thematic from asset class name
-    class_name = asset.__class__.__name__.lower()
     thematic_mappings = {
         'naturerwiz': 'water',
         'naturerwiasset': 'water',
@@ -317,8 +412,60 @@ def _get_asset_thematic(asset):
         'parking': 'parking',
         'parkingspot': 'parking',
         'parkinglocation': 'parking',
+        'clean': 'clean',
+        'cleanlocation': 'clean',
+        'cleanlitterbin': 'clean',
+        'cleanglasscontainer': 'clean',
     }
+
+    if isinstance(asset, type):
+        class_name = asset.__name__.lower()
+        for prefix, thematic_code in thematic_mappings.items():
+            if class_name.startswith(prefix):
+                try:
+                    return Thematic.objects.get(code=thematic_code)
+                except Thematic.DoesNotExist:
+                    pass
+        # Fallback: check AssetCategory via ContentType
+        from django.contrib.contenttypes.models import ContentType
+        from assets.models import AssetCategory
+        ct = ContentType.objects.filter(model=class_name).first()
+        if ct:
+            cat = AssetCategory.objects.filter(allowed_models=ct).first()
+            if cat and cat.thematic:
+                return cat.thematic
+        return None
+
+    # Try from category
+    try:
+        if hasattr(asset, 'category') and asset.category and hasattr(asset.category, 'thematic'):
+            return asset.category.thematic
+    except Exception:
+        pass
+
+    # Try direct thematic attribute
+    try:
+        if hasattr(asset, 'thematic') and asset.thematic and hasattr(asset.thematic, 'code'):
+            return asset.thematic
+    except Exception:
+        pass
+
+    # Try from parent building category
+    try:
+        if hasattr(asset, 'building') and asset.building and hasattr(asset.building, 'category') and asset.building.category:
+            return asset.building.category.thematic
+    except Exception:
+        pass
     
+    # Try from model's category
+    try:
+        if hasattr(asset, 'model') and asset.model and hasattr(asset.model, 'category') and asset.model.category:
+            return asset.model.category.thematic
+    except Exception:
+        pass
+    
+    # Fallback: deduce thematic from asset class name
+    class_name = asset.__class__.__name__.lower()
     for prefix, thematic_code in thematic_mappings.items():
         if class_name.startswith(prefix):
             try:

loko/assets/templatetags/asset_tags.py

@@ -590,20 +590,26 @@ def asset_inspections(context, asset):
 
     from django.contrib.contenttypes.models import ContentType
     from inspections.models import Inspection
+    from inspections.permissions import filter_viewable_inspections_for_user
     from assets.permissions import can_inspect_asset
 
     from inspections.registry import get_inspection_partial_template
 
-    asset_ct = ContentType.objects.get_for_model(asset)
-    inspections = list(Inspection.objects.filter(
-        asset_content_type=asset_ct,
-        asset_object_id=asset.pk
-    ).select_related('inspector').prefetch_related('documents').order_by('-inspection_date'))
-
     request = context.get('request')
     user = request.user if request else None
-    can_insp = False
 
+    asset_ct = ContentType.objects.get_for_model(asset)
+    base_qs = Inspection.objects.filter(
+        asset_content_type=asset_ct,
+        asset_object_id=asset.pk
+    ).select_related('inspector').prefetch_related('documents').order_by('-inspection_date')
+
+    if user and user.is_authenticated:
+        inspections = list(filter_viewable_inspections_for_user(user, base_qs))
+    else:
+        inspections = []
+
+    can_insp = False
     if user and user.is_authenticated:
         can_insp = can_inspect_asset(user, asset)
 

loko/assets/views/parking.py

@@ -344,9 +344,13 @@ def parking_assets_detail(request, asset_model, asset_id):
     content_type = ContentType.objects.get_for_model(ParkingSpot)
 
     from inspections.models import Inspection
-    inspections = Inspection.objects.filter(
-        asset_content_type=content_type,
-        asset_object_id=spot.id
+    from inspections.permissions import filter_viewable_inspections_for_user
+    inspections = filter_viewable_inspections_for_user(
+        request.user,
+        Inspection.objects.filter(
+            asset_content_type=content_type,
+            asset_object_id=spot.id
+        )
     ).order_by('-inspection_date')
 
     from interventions.models import Intervention

loko/inspections/apps.py

@@ -12,9 +12,15 @@ class InspectionsConfig(AppConfig):
             from common.private_files.registry import register
 
             def guard_inspection_document(obj, user):
-                if not user.is_authenticated:
+                if not user or not user.is_authenticated:
                     return False
-                return True
+                if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
+                    return True
+                user_config = getattr(user, 'config', None)
+                if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern):
+                    return True
+                from inspections.permissions import can_view_inspection
+                return can_view_inspection(user, getattr(obj, 'inspection', None))
 
             register('inspections', 'InspectionDocument', 'file', guard_inspection_document)
             register('assets', 'InspectionDocument', 'file', guard_inspection_document)

loko/inspections/permissions.py

@@ -1,4 +1,23 @@
+from functools import reduce
+from operator import or_
+from collections import defaultdict
 from django.conf import settings
+from django.db.models import Q
+from django.contrib.contenttypes.models import ContentType
+
+
+def is_user_internal_or_admin(user):
+    """
+    Détermine si un utilisateur est un utilisateur interne ou un administrateur ayant un accès global.
+    """
+    if not user or not user.is_authenticated:
+        return False
+    if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
+        return True
+    user_config = getattr(user, 'config', None)
+    if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern):
+        return True
+    return False
 
 
 def can_edit_inspection(user, inspection):
@@ -13,7 +32,7 @@ def can_edit_inspection(user, inspection):
     if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
         return True
 
-    # Administrateur au sens configuration de rôle StreetUp
+    # Administrateur au sens configuration de rôle
     user_config = getattr(user, 'config', None)
     if user_config and hasattr(user_config, 'has_role'):
         if user_config.has_role('admin') or user_config.has_role('top_manager'):
@@ -24,3 +43,262 @@ def can_edit_inspection(user, inspection):
         return True
 
     return False
+
+
+def can_view_inspection(user, inspection):
+    """
+    Vérifie si un utilisateur a le droit de visualiser une inspection.
+    - Administrateurs et utilisateurs internes (is_intern=True) : accès complet.
+    - Utilisateurs externes : seules les inspections qui les concernent sont visibles :
+      1. L'utilisateur est l'auteur/inspecteur de l'inspection.
+      2. L'inspection est liée à une mission d'intervention visible par l'utilisateur.
+      3. L'inspection est liée à une intervention contrôlée visible par l'utilisateur.
+      4. L'inspection est liée à un asset ou une intervention cible visible par l'utilisateur.
+    """
+    if not user or not user.is_authenticated:
+        return False
+
+    if is_user_internal_or_admin(user):
+        return True
+
+    if not inspection:
+        return False
+
+    # 1. Auteur / Inspecteur de l'inspection
+    if inspection.inspector_id == user.id:
+        return True
+
+    # 2. Mission d'intervention
+    if inspection.mission_intervention_id:
+        from interventions.permissions import can_view_intervention
+        if can_view_intervention(user, inspection.mission_intervention):
+            return True
+
+    # 3. Intervention de travaux contrôlée
+    if inspection.checked_intervention_id:
+        from interventions.permissions import can_view_intervention
+        if can_view_intervention(user, inspection.checked_intervention):
+            return True
+
+    # 4. Cible (Asset ou Intervention)
+    if inspection.asset_content_type_id and inspection.asset_object_id:
+        if inspection.asset_content_type.app_label == 'interventions':
+            from interventions.permissions import can_view_intervention
+            target = inspection.asset
+            if target and can_view_intervention(user, target):
+                return True
+        else:
+            from assets.permissions import can_view_asset
+            target = inspection.asset
+            if target and can_view_asset(user, target):
+                return True
+
+    return False
+
+
+def get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids):
+    """
+    Retourne un QuerySet filtré des instances de ModelClass liées aux contrats autorisés,
+    ou None si ModelClass n'est pas un modèle gérant des contrats.
+    """
+    class_name = ModelClass.__name__.lower()
+    
+    # Structures
+    if class_name == 'structure':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(contracts__status='active', contracts__contract_id__in=accessible_contract_ids)
+    elif class_name == 'structurelocation':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(structure__contracts__status='active', structure__contracts__contract_id__in=accessible_contract_ids)
+    elif class_name == 'structuregeoasset':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(location__structure__contracts__status='active', location__structure__contracts__contract_id__in=accessible_contract_ids)
+
+    # ITS
+    elif class_name == 'itslocation':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(its_locations_contracts__status='active', its_locations_contracts__contract_id__in=accessible_contract_ids)
+    elif class_name == 'itsgeoasset':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(its_assets_contracts__status='active', its_assets_contracts__contract_id__in=accessible_contract_ids)
+
+    # Control Centers
+    elif class_name == 'controlcenter':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(controlcenters_contracts__status='active', controlcenters_contracts__contract_id__in=accessible_contract_ids)
+    elif class_name == 'controlcenterlocation':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(control_center__controlcenters_contracts__status='active', control_center__controlcenters_contracts__contract_id__in=accessible_contract_ids)
+    elif class_name == 'controlcentergeoasset':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(controlcenters_assets_contracts__status='active', controlcenters_assets_contracts__contract_id__in=accessible_contract_ids)
+
+    # Traffic Lights
+    elif class_name == 'trafficlightintersection':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(trafficlight_contracts__status='active', trafficlight_contracts__contract_id__in=accessible_contract_ids)
+    elif class_name in ('trafficlightpole', 'trafficlightlantern', 'trafficlightcable', 'trafficlightdetector',
+                         'trafficlightcontroller', 'trafficlightelectricalcabinet', 'trafficlightradar',
+                         'trafficlightaccessory'):
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(intersection__trafficlight_contracts__status='active', intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids)
+    elif class_name == 'trafficlightradarpole':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(pole__intersection__trafficlight_contracts__status='active', pole__intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids)
+
+    # Clean
+    elif class_name == 'cleanlocation':
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(clean_location_contracts__contract_id__in=accessible_contract_ids)
+    elif class_name in ('cleanlitterbin', 'cleanglasscontainer'):
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(location__clean_location_contracts__contract_id__in=accessible_contract_ids)
+
+    # Artworks
+    elif class_name in ('artwork', 'monument', 'fountain'):
+        if not accessible_contract_ids:
+            return ModelClass.objects.none()
+        return ModelClass.objects.filter(maintenance_contract_id__in=accessible_contract_ids)
+
+    # Modèles sans notion de contrat (ex: ParkingSpot, NatureRWIZ, Road, PublicLighting, Sign, etc.)
+    return None
+
+
+def filter_viewable_inspections_for_user(user, qs=None):
+    """
+    Filtre un QuerySet d'inspections pour ne renvoyer que celles visibles par l'utilisateur.
+    Pour les internes et administrateurs : toutes les inspections (qs complet).
+    Pour les externes : seules les inspections qui les concernent :
+    - Inspections créées par l'utilisateur (inspecteur)
+    - Inspections liées à une mission d'intervention ou intervention contrôlée visible
+    - Inspections dont l'asset ou intervention cible est visible
+    """
+    from inspections.models import Inspection
+
+    if qs is None:
+        qs = Inspection.objects.all()
+
+    if not user or not user.is_authenticated:
+        return qs.none()
+
+    if is_user_internal_or_admin(user):
+        return qs
+
+    user_config = getattr(user, 'config', None)
+    conditions = []
+
+    # 1. Auteur / Inspecteur de l'inspection
+    conditions.append(Q(inspector=user))
+
+    # 2. Interventions liées visibles (missions d'intervention, contrôles post-travaux, ou cible intervention)
+    from interventions.permissions import filter_viewable_interventions_for_user
+    viewable_interventions = filter_viewable_interventions_for_user(user)
+
+    conditions.append(Q(mission_intervention__in=viewable_interventions))
+    conditions.append(Q(checked_intervention__in=viewable_interventions))
+
+    interv_ct = ContentType.objects.filter(app_label='interventions', model='intervention').first()
+    if interv_ct:
+        conditions.append(Q(asset_content_type=interv_ct, asset_object_id__in=viewable_interventions.values('pk')))
+
+    # 3. Assets cibles visibles
+    if user_config:
+        from common.models import UserThematics, UserContractAccess, UserAssetAccess, UserAssetTypeAccess, AssetTypePermissionConfig
+        from assets.permissions import _get_asset_thematic
+
+        viewable_thematic_ids = set(
+            UserThematics.objects.filter(user_config=user_config, can_view_assets=True).values_list('thematic_id', flat=True)
+        )
+        must_limit = not user_config.is_intern and user_config.limit_assets_to_contracts
+        accessible_contract_ids = list(
+            UserContractAccess.objects.filter(user_config=user_config, can_view_assets=True).values_list('contract_id', flat=True)
+        ) if must_limit else []
+
+        explicit_denies = defaultdict(set)
+        for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=False).values_list('content_type_id', 'object_id'):
+            explicit_denies[row[0]].add(row[1])
+
+        explicit_allows = defaultdict(set)
+        for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=True).values_list('content_type_id', 'object_id'):
+            explicit_allows[row[0]].add(row[1])
+
+        # ContentTypes présents dans les inspections ciblées
+        ct_ids = set(qs.values_list('asset_content_type_id', flat=True).distinct())
+
+        for ct_id in ct_ids:
+            if not ct_id:
+                continue
+            try:
+                ct = ContentType.objects.get(pk=ct_id)
+            except ContentType.DoesNotExist:
+                continue
+
+            if ct.app_label == 'interventions':
+                continue
+
+            ModelClass = ct.model_class()
+            if not ModelClass:
+                continue
+
+            # Vérifier si AssetTypePermissionConfig exige des permissions explicites
+            requires_explicit = AssetTypePermissionConfig.objects.filter(content_type=ct, requires_explicit_permissions=True).exists()
+            type_access = UserAssetTypeAccess.objects.filter(user_config=user_config, content_type=ct).first()
+
+            if type_access and not type_access.can_view:
+                allowed_ids = explicit_allows.get(ct.id, set())
+                if allowed_ids:
+                    conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids))
+                continue
+
+            if requires_explicit:
+                allowed_ids = explicit_allows.get(ct.id, set())
+                if allowed_ids:
+                    conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids))
+                continue
+
+            # Vérifier l'accès thématique
+            th = _get_asset_thematic(ModelClass)
+            if th and th.id not in viewable_thematic_ids:
+                allowed_ids = explicit_allows.get(ct.id, set())
+                if allowed_ids:
+                    conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids))
+                continue
+
+            # Thématique autorisée (ou sans thématique assignée)
+            if must_limit:
+                filtered_asset_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids)
+                if filtered_asset_qs is not None:
+                    ct_q = Q(asset_content_type=ct, asset_object_id__in=filtered_asset_qs.values('pk'))
+                else:
+                    ct_q = Q(asset_content_type=ct)
+            else:
+                ct_q = Q(asset_content_type=ct)
+
+            denied_ids = explicit_denies.get(ct.id, set())
+            if denied_ids:
+                ct_q &= ~Q(asset_object_id__in=denied_ids)
+
+            allowed_ids = explicit_allows.get(ct.id, set())
+            if allowed_ids:
+                ct_q |= Q(asset_content_type=ct, asset_object_id__in=allowed_ids)
+
+            conditions.append(ct_q)
+
+    if not conditions:
+        return qs.none()
+
+    q = reduce(or_, conditions)
+    return qs.filter(q).distinct()

loko/inspections/tests.py

@@ -496,3 +496,341 @@ class InspectionModelAndViewsTestCase(TestCase):
         self.assertIsNone(self.spot.last_inspection_date)
 
 
+class ExternalUserInspectionVisibilityTestCase(TestCase):
+    def setUp(self):
+        self.client = Client()
+
+        # Utilisateur interne
+        self.internal_user = User.objects.create_user(
+            username='intern_user',
+            password='password123',
+            first_name='Intern',
+            last_name='User'
+        )
+        from common.models import UserConfig, Role, UserThematics, UserContractAccess
+        from contracts.models import Contract
+        from assets.models import Structure, StructureContract, AssetCategory
+
+        config_intern = UserConfig.objects.create(
+            user=self.internal_user,
+            is_intern=True,
+            limit_assets_to_contracts=False
+        )
+
+        # Utilisateur externe
+        self.external_user = User.objects.create_user(
+            username='extern_user',
+            password='password123',
+            first_name='Extern',
+            last_name='User'
+        )
+        self.external_config = UserConfig.objects.create(
+            user=self.external_user,
+            is_intern=False,
+            limit_assets_to_contracts=True
+        )
+        role_ext, _ = Role.objects.get_or_create(name='external_manager')
+        self.external_config.roles.add(role_ext)
+
+        # Thématiques
+        self.thematic_parking, _ = Thematic.objects.get_or_create(
+            code='parking',
+            defaults={'name_fr': 'Stationnement'}
+        )
+        self.thematic_water, _ = Thematic.objects.get_or_create(
+            code='water',
+            defaults={'name_fr': 'Gestion de l\'Eau'}
+        )
+        self.thematic_structures, _ = Thematic.objects.get_or_create(
+            code='structures',
+            defaults={'name_fr': 'Ouvrages d\'art'}
+        )
+
+        # L'externe a accès aux assets Parking et Structures, mais PAS à Water
+        UserThematics.objects.create(
+            user_config=self.external_config,
+            thematic=self.thematic_parking,
+            can_view_assets=True,
+            can_view_interventions=True
+        )
+        UserThematics.objects.create(
+            user_config=self.external_config,
+            thematic=self.thematic_structures,
+            can_view_assets=True,
+            can_view_interventions=True
+        )
+        UserThematics.objects.create(
+            user_config=self.external_config,
+            thematic=self.thematic_water,
+            can_view_assets=False,
+            can_view_interventions=False
+        )
+
+        # Contrats
+        from contracts.models import Company, Contract
+        self.company = Company.objects.create(name="Company Ext Test")
+        today = timezone.now().date()
+        self.contract_a = Contract.objects.create(
+            company=self.company,
+            contract_number="CTR-EXT-A",
+            start_date=today,
+            end_date=today + timezone.timedelta(days=365),
+            is_active=True
+        )
+        self.contract_a.thematics.add(self.thematic_structures)
+
+        self.contract_b = Contract.objects.create(
+            company=self.company,
+            contract_number="CTR-EXT-B",
+            start_date=today,
+            end_date=today + timezone.timedelta(days=365),
+            is_active=True
+        )
+        self.contract_b.thematics.add(self.thematic_structures)
+
+        # L'externe a accès au contrat A mais pas au B
+        uca_a = UserContractAccess.objects.create(
+            user_config=self.external_config,
+            contract=self.contract_a,
+            can_view_assets=True,
+            can_view_interventions=True
+        )
+        from common.models import UserContractStatusPermission
+        UserContractStatusPermission.objects.create(
+            user_contract=uca_a,
+            status='in_progress',
+            can_view=True
+        )
+
+        # Assets
+        self.spot = ParkingSpot.objects.create(name_fr="Emplacement Ext")
+        self.spot_ct = ContentType.objects.get_for_model(ParkingSpot)
+
+        self.rwiz = NatureRWIZ.objects.create(name_fr="Bassin Water")
+        self.rwiz_ct = ContentType.objects.get_for_model(NatureRWIZ)
+
+        self.struct_a = Structure.objects.create(code="STR-A", name_fr="Pont A")
+        StructureContract.objects.create(structure=self.struct_a, contract=self.contract_a, status='active')
+        self.struct_ct = ContentType.objects.get_for_model(Structure)
+
+        self.struct_b = Structure.objects.create(code="STR-B", name_fr="Pont B")
+        StructureContract.objects.create(structure=self.struct_b, contract=self.contract_b, status='active')
+
+        # Interventions
+        self.interv_a = Intervention.objects.create(
+            title="Intervention Contrat A",
+            contract=self.contract_a,
+            thematic=self.thematic_structures,
+            status="in_progress"
+        )
+        self.interv_b = Intervention.objects.create(
+            title="Intervention Contrat B",
+            contract=self.contract_b,
+            thematic=self.thematic_structures,
+            status="in_progress"
+        )
+
+    def test_external_user_thematic_filtering(self):
+        from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user
+
+        insp_parking = Inspection.objects.create(
+            asset_content_type=self.spot_ct,
+            asset_object_id=self.spot.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+        insp_water = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+
+        # L'externe peut voir le parking (can_view_assets=True sur parking)
+        self.assertTrue(can_view_inspection(self.external_user, insp_parking))
+        # L'externe ne peut pas voir water (can_view_assets=False sur water)
+        self.assertFalse(can_view_inspection(self.external_user, insp_water))
+
+        # QuerySet filtering
+        qs = filter_viewable_inspections_for_user(self.external_user)
+        self.assertIn(insp_parking, qs)
+        self.assertNotIn(insp_water, qs)
+
+    def test_external_user_contract_filtering(self):
+        from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user
+
+        insp_struct_a = Inspection.objects.create(
+            asset_content_type=self.struct_ct,
+            asset_object_id=self.struct_a.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+        insp_struct_b = Inspection.objects.create(
+            asset_content_type=self.struct_ct,
+            asset_object_id=self.struct_b.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+
+        # L'externe a accès au contrat A -> voit insp_struct_a
+        self.assertTrue(can_view_inspection(self.external_user, insp_struct_a))
+        # L'externe n'a pas accès au contrat B -> ne voit pas insp_struct_b
+        self.assertFalse(can_view_inspection(self.external_user, insp_struct_b))
+
+        # QuerySet filtering
+        qs = filter_viewable_inspections_for_user(self.external_user)
+        self.assertIn(insp_struct_a, qs)
+        self.assertNotIn(insp_struct_b, qs)
+
+    def test_external_user_intervention_linked_inspection(self):
+        from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user
+
+        # Inspection sur asset d'une thématique non autorisée, mais liée à une intervention autorisée (mission)
+        insp_mission = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.internal_user,
+            mission_intervention=self.interv_a,
+            result_status='compliant'
+        )
+        # Inspection liée à une intervention non autorisée
+        insp_mission_unauth = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.internal_user,
+            mission_intervention=self.interv_b,
+            result_status='compliant'
+        )
+
+        self.assertTrue(can_view_inspection(self.external_user, insp_mission))
+        self.assertFalse(can_view_inspection(self.external_user, insp_mission_unauth))
+
+        qs = filter_viewable_inspections_for_user(self.external_user)
+        self.assertIn(insp_mission, qs)
+        self.assertNotIn(insp_mission_unauth, qs)
+
+    def test_external_user_is_inspector_always_visible(self):
+        from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user
+
+        # Inspection réalisée par l'externe lui-même sur une thématique normalement restreinte
+        insp_by_extern = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.external_user,
+            result_status='compliant'
+        )
+
+        self.assertTrue(can_view_inspection(self.external_user, insp_by_extern))
+        qs = filter_viewable_inspections_for_user(self.external_user)
+        self.assertIn(insp_by_extern, qs)
+
+    def test_external_user_inspections_list_view(self):
+        insp_parking = Inspection.objects.create(
+            asset_content_type=self.spot_ct,
+            asset_object_id=self.spot.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+        insp_water = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+
+        self.client.login(username='extern_user', password='password123')
+        url = reverse('inspections:inspections_list')
+        response = self.client.get(url, {'tab': 'all'})
+        self.assertEqual(response.status_code, 200)
+
+        inspections = list(response.context['inspections'])
+        self.assertIn(insp_parking, inspections)
+        self.assertNotIn(insp_water, inspections)
+
+    def test_external_user_inspection_detail_api(self):
+        insp_parking = Inspection.objects.create(
+            asset_content_type=self.spot_ct,
+            asset_object_id=self.spot.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+        insp_water = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+
+        self.client.login(username='extern_user', password='password123')
+
+        # Inspection autorisée -> 200
+        url_ok = reverse('inspections:get_inspection_detail_api', kwargs={'inspection_id': insp_parking.id})
+        res_ok = self.client.get(url_ok)
+        self.assertEqual(res_ok.status_code, 200)
+        self.assertTrue(res_ok.json()['success'])
+
+        # Inspection non autorisée -> 403
+        url_forbidden = reverse('inspections:get_inspection_detail_api', kwargs={'inspection_id': insp_water.id})
+        res_forbidden = self.client.get(url_forbidden)
+        self.assertEqual(res_forbidden.status_code, 403)
+        self.assertFalse(res_forbidden.json()['success'])
+
+    def test_guard_inspection_document(self):
+        from inspections.models import InspectionDocument
+        from django.core.files.uploadedfile import SimpleUploadedFile
+        from inspections.apps import InspectionsConfig
+
+        insp_parking = Inspection.objects.create(
+            asset_content_type=self.spot_ct,
+            asset_object_id=self.spot.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+        insp_water = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.internal_user,
+            result_status='compliant'
+        )
+
+        dummy_file = SimpleUploadedFile("doc.txt", b"content", content_type="text/plain")
+        doc_parking = InspectionDocument.objects.create(inspection=insp_parking, file=dummy_file)
+        doc_water = InspectionDocument.objects.create(inspection=insp_water, file=dummy_file)
+
+        from inspections.permissions import can_view_inspection
+        self.assertTrue(can_view_inspection(self.external_user, doc_parking.inspection))
+        self.assertFalse(can_view_inspection(self.external_user, doc_water.inspection))
+
+    def test_internal_user_sees_all_inspections(self):
+        from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user
+
+        insp_parking = Inspection.objects.create(
+            asset_content_type=self.spot_ct,
+            asset_object_id=self.spot.id,
+            inspector=self.external_user,
+            result_status='compliant'
+        )
+        insp_water = Inspection.objects.create(
+            asset_content_type=self.rwiz_ct,
+            asset_object_id=self.rwiz.id,
+            inspector=self.external_user,
+            result_status='compliant'
+        )
+        insp_struct_b = Inspection.objects.create(
+            asset_content_type=self.struct_ct,
+            asset_object_id=self.struct_b.id,
+            inspector=self.external_user,
+            result_status='compliant'
+        )
+
+        self.assertTrue(can_view_inspection(self.internal_user, insp_parking))
+        self.assertTrue(can_view_inspection(self.internal_user, insp_water))
+        self.assertTrue(can_view_inspection(self.internal_user, insp_struct_b))
+
+        qs = filter_viewable_inspections_for_user(self.internal_user)
+        self.assertIn(insp_parking, qs)
+        self.assertIn(insp_water, qs)
+        self.assertIn(insp_struct_b, qs)
+
+
+

loko/inspections/views.py

@@ -49,7 +49,9 @@ def inspections_list(request):
     checked_interv_id = request.GET.get('checked_interv_id', '')
     search_q = request.GET.get('q', '').strip()
 
-    qs = Inspection.objects.all().select_related(
+    from .permissions import filter_viewable_inspections_for_user
+    qs = filter_viewable_inspections_for_user(request.user)
+    qs = qs.select_related(
         'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention'
     ).prefetch_related('documents').order_by('-inspection_date')
 
@@ -433,10 +435,11 @@ def get_inspection_detail_api(request, inspection_id):
     inspection_obj = get_object_or_404(Inspection, pk=inspection_id)
     leaf = inspection_obj.detailed_inspection
 
-    if not leaf.can_edit(request.user):
+    from .permissions import can_view_inspection
+    if not can_view_inspection(request.user, leaf):
         return JsonResponse({
             'success': False,
-            'error': str(_("Permission refusée. Seuls l'auteur et les administrateurs peuvent modifier cette inspection."))
+            'error': str(_("Permission refusée. Vous n'avez pas accès à cette inspection."))
         }, status=403)
 
     docs_data = []
@@ -665,7 +668,9 @@ def inspections_geojson(request):
     if period == 'disabled':
         return JsonResponse({'type': 'FeatureCollection', 'features': []})
 
-    qs = Inspection.objects.all().select_related('asset_content_type', 'inspector', 'mission_intervention').prefetch_related('documents')
+    from .permissions import filter_viewable_inspections_for_user, get_contract_filtered_asset_qs
+    qs = filter_viewable_inspections_for_user(request.user)
+    qs = qs.select_related('asset_content_type', 'inspector', 'mission_intervention').prefetch_related('documents')
 
     if mission_id:
         qs = qs.filter(mission_intervention_id=mission_id)
@@ -710,6 +715,19 @@ def inspections_geojson(request):
 
     features = []
 
+    user_config = getattr(request.user, 'config', None)
+    is_external = (user_config is None) or (not user_config.is_intern)
+    must_limit = is_external and (not user_config or user_config.limit_assets_to_contracts)
+    accessible_contract_ids = []
+    if must_limit and user_config:
+        from common.models import UserContractAccess
+        accessible_contract_ids = list(
+            UserContractAccess.objects.filter(
+                user_config=user_config,
+                can_view_assets=True
+            ).values_list('contract_id', flat=True)
+        )
+
     for ct_id in ct_ids:
         if not ct_id:
             continue
@@ -720,6 +738,11 @@ def inspections_geojson(request):
                 continue
 
             asset_qs = ModelClass.objects.filter(geom__isnull=False)
+            if must_limit:
+                contract_filtered_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids)
+                if contract_filtered_qs is not None:
+                    asset_qs = asset_qs.filter(pk__in=contract_filtered_qs.values('pk'))
+
             if bbox_geom:
                 bbox_geom_3812 = bbox_geom.transform(3812, clone=True)
                 asset_qs = asset_qs.filter(geom__intersects=bbox_geom_3812)