558 lines
19 KiB
Python
558 lines
19 KiB
Python
# --------------------
|
|
# Asset Permissions
|
|
# --------------------
|
|
|
|
from functools import wraps
|
|
from django.shortcuts import get_object_or_404
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.http import HttpResponseForbidden, JsonResponse
|
|
from django.utils.translation import gettext as _
|
|
|
|
from common.models import UserConfig, UserThematics, UserAssetAccess, UserAssetTypeAccess, AssetTypePermissionConfig
|
|
|
|
|
|
# Fields editable by role for assets
|
|
ASSET_EDITABLE_FIELDS_BY_ROLE = {
|
|
'admin': [
|
|
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
'manager': [
|
|
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
'controller': [
|
|
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
'external_manager': [
|
|
'name_fr', 'name_nl', 'category', 'brand', 'serial_number'
|
|
],
|
|
'technician': ['name_fr', 'name_nl', 'category', 'brand', 'serial_number'],
|
|
'operator': [],
|
|
'observer': [],
|
|
'viewer': [],
|
|
'editor': [
|
|
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
}
|
|
|
|
# Actions allowed by role for assets
|
|
ASSET_ACTIONS_BY_ROLE = {
|
|
'admin': ['edit', 'create', 'archive', 'replace', 'bulk_archive'],
|
|
'manager': ['edit', 'create', 'archive', 'replace', 'bulk_archive'],
|
|
'controller': ['edit', 'create', 'archive', 'replace', 'bulk_archive'],
|
|
'external_manager': ['edit', 'create', 'replace'],
|
|
'operator': [],
|
|
'technician': ['edit', 'create', 'replace'],
|
|
'observer': [],
|
|
'viewer': [],
|
|
'editor': ['edit', 'create', 'archive', 'replace'],
|
|
}
|
|
|
|
|
|
def get_user_asset_thematic_access(user, thematic):
|
|
"""
|
|
Check if user has access to view/edit assets for a given thematic.
|
|
Returns a tuple (can_view, can_edit).
|
|
"""
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False, False
|
|
|
|
try:
|
|
user_thematic = UserThematics.objects.get(
|
|
user_config=user_config,
|
|
thematic=thematic
|
|
)
|
|
return user_thematic.can_view_assets, user_thematic.can_edit_assets
|
|
except UserThematics.DoesNotExist:
|
|
return False, False
|
|
|
|
|
|
def _check_instance_permission(user, obj, permission_field, thematic_fallback_fn=None):
|
|
"""
|
|
Shared logic for per-instance permission checks (assets and locations).
|
|
|
|
Priority order (mirrors contract permission logic):
|
|
1. Admin role → always True.
|
|
2. Explicit UserAssetAccess row exists → use its value (True or False, no fallback).
|
|
3. No row exists:
|
|
a. AssetTypePermissionConfig.requires_explicit_permissions=True → False.
|
|
b. Otherwise → call thematic_fallback_fn() (defaults to False if None).
|
|
|
|
Args:
|
|
user: Django User instance.
|
|
obj: Asset or Location model instance.
|
|
permission_field: 'can_view', 'can_edit', or 'can_delete'.
|
|
thematic_fallback_fn: callable() → bool; used when no explicit row exists
|
|
and the type does not require explicit permissions.
|
|
"""
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
# Step 1: admin bypass
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False
|
|
|
|
if user_config.roles.filter(name='admin').exists():
|
|
return True
|
|
|
|
# Step 2: explicit per-instance row
|
|
ct = ContentType.objects.get_for_model(obj)
|
|
try:
|
|
access = UserAssetAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
object_id=obj.pk,
|
|
)
|
|
return getattr(access, permission_field, False)
|
|
except UserAssetAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Step 3: per-user type-level access (applies to all instances of this content type)
|
|
try:
|
|
type_access = UserAssetTypeAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
)
|
|
return getattr(type_access, permission_field, False)
|
|
except UserAssetTypeAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Step 4a: type-level restriction
|
|
try:
|
|
type_config = AssetTypePermissionConfig.objects.get(content_type=ct)
|
|
if type_config.requires_explicit_permissions:
|
|
return False
|
|
except AssetTypePermissionConfig.DoesNotExist:
|
|
pass
|
|
|
|
# Step 4b: thematic fallback
|
|
if thematic_fallback_fn is not None:
|
|
return thematic_fallback_fn()
|
|
return False
|
|
|
|
|
|
def can_view_asset(user, asset):
|
|
"""Check if user can view a specific asset."""
|
|
def _fallback():
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
can_view, _ = get_user_asset_thematic_access(user, thematic)
|
|
return can_view
|
|
|
|
return _check_instance_permission(user, asset, 'can_view', _fallback)
|
|
|
|
|
|
def can_edit_asset(user, asset):
|
|
"""Check if user can edit a specific asset."""
|
|
def _fallback():
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
_, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
return can_edit
|
|
|
|
return _check_instance_permission(user, asset, 'can_edit', _fallback)
|
|
|
|
|
|
def can_delete_asset(user, asset):
|
|
"""Check if user can delete (archive/remove) a specific asset."""
|
|
def _fallback():
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
_, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
# By default, delete follows edit rights at the thematic level
|
|
return can_edit
|
|
|
|
return _check_instance_permission(user, asset, 'can_delete', _fallback)
|
|
|
|
|
|
def can_validate_asset(user, asset):
|
|
"""Check if user has permission to validate a specific asset."""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False
|
|
|
|
if user_config.roles.filter(name='admin').exists():
|
|
return True
|
|
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
|
|
try:
|
|
ut = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return ut.can_validate_assets
|
|
except UserThematics.DoesNotExist:
|
|
return False
|
|
|
|
|
|
def can_view_location(user, location):
|
|
"""Check if user can view a specific location."""
|
|
def _fallback():
|
|
thematic = _get_location_thematic(location)
|
|
if not thematic:
|
|
return False
|
|
can_view, _ = get_user_asset_thematic_access(user, thematic)
|
|
return can_view
|
|
|
|
return _check_instance_permission(user, location, 'can_view', _fallback)
|
|
|
|
|
|
def can_edit_location(user, location):
|
|
"""Check if user can edit a specific location."""
|
|
def _fallback():
|
|
thematic = _get_location_thematic(location)
|
|
if not thematic:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
ut = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return ut.can_edit_locations
|
|
except Exception:
|
|
return False
|
|
|
|
return _check_instance_permission(user, location, 'can_edit', _fallback)
|
|
|
|
|
|
def can_delete_location(user, location):
|
|
"""Check if user can delete a specific location."""
|
|
def _fallback():
|
|
thematic = _get_location_thematic(location)
|
|
if not thematic:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
ut = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return ut.can_edit_locations
|
|
except Exception:
|
|
return False
|
|
|
|
return _check_instance_permission(user, location, 'can_delete', _fallback)
|
|
|
|
|
|
def _get_asset_thematic(asset):
|
|
"""Get the thematic for an asset."""
|
|
from common.models import Thematic
|
|
|
|
# Try from category
|
|
if hasattr(asset, 'category') and asset.category:
|
|
return asset.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 = {
|
|
'naturerwiasset': 'water',
|
|
'trafficlight': 'trafficlights',
|
|
'publiclighting': 'publiclighting',
|
|
'road': 'roads',
|
|
'artwork': 'artworks',
|
|
'monument': 'artworks',
|
|
'fountain': 'artworks',
|
|
'structure': 'structures',
|
|
'nature': 'nature',
|
|
'green': 'nature',
|
|
'sign': 'sign',
|
|
'controlcenter': 'controlcenters',
|
|
'its': 'its',
|
|
}
|
|
|
|
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
|
|
|
|
return None
|
|
|
|
|
|
def _get_location_thematic(location):
|
|
"""Deduce the thematic for a location instance."""
|
|
from common.models import Thematic
|
|
|
|
# RoadStreet, NatureLocation, etc. have a `thematic` attribute via their linked assets/categories,
|
|
# but locations are usually tied to a thematic via class name.
|
|
class_name = location.__class__.__name__.lower()
|
|
thematic_mappings = {
|
|
'naturerwiz': 'water',
|
|
'roadstreet': 'roads',
|
|
'trafficlightintersection': 'trafficlights',
|
|
'naturelocation': 'nature',
|
|
'structurelocation': 'structures',
|
|
'controlcenter': 'controlcenters',
|
|
'itslocation': 'its',
|
|
'cleanlocation': 'clean',
|
|
'publiclightingstreet': 'publiclighting',
|
|
'signstreet': 'sign',
|
|
}
|
|
thematic_code = thematic_mappings.get(class_name)
|
|
if not thematic_code:
|
|
for prefix, code in thematic_mappings.items():
|
|
if class_name.startswith(prefix[:5]):
|
|
thematic_code = code
|
|
break
|
|
if not thematic_code:
|
|
return None
|
|
try:
|
|
return Thematic.objects.get(code=thematic_code)
|
|
except Thematic.DoesNotExist:
|
|
return None
|
|
|
|
|
|
def get_allowed_update_fields_for_asset(user, asset):
|
|
"""
|
|
Get the list of fields the user is allowed to update for a specific asset.
|
|
|
|
Priority:
|
|
1. If a UserAssetAccess row with non-empty editable_fields exists, that list is returned
|
|
as-is (completely replaces role-based logic).
|
|
2. Otherwise uses the existing role-based logic (ASSET_EDITABLE_FIELDS_BY_ROLE).
|
|
"""
|
|
if not can_edit_asset(user, asset):
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
# Check for per-instance editable_fields override
|
|
from django.contrib.contenttypes.models import ContentType
|
|
ct = ContentType.objects.get_for_model(asset)
|
|
try:
|
|
access = UserAssetAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
object_id=asset.pk,
|
|
)
|
|
if access.editable_fields: # non-null, non-empty list → use it
|
|
return list(access.editable_fields)
|
|
except UserAssetAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Check for per-type editable_fields override (applies to all instances)
|
|
try:
|
|
type_access = UserAssetTypeAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
)
|
|
if type_access.editable_fields: # non-null, non-empty list → use it
|
|
return list(type_access.editable_fields)
|
|
except UserAssetTypeAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Fall back to role-based logic
|
|
allowed_fields = set()
|
|
for role in user_config.roles.all():
|
|
role_fields = ASSET_EDITABLE_FIELDS_BY_ROLE.get(role.name, [])
|
|
allowed_fields.update(role_fields)
|
|
|
|
# Check validation permission
|
|
if can_validate_asset(user, asset):
|
|
allowed_fields.add('validation_status')
|
|
|
|
# Also add asset-specific fields (not in base models) when the user already has some edit rights
|
|
if allowed_fields and asset is not None:
|
|
from django.db.models import ForeignKey, OneToOneField, ManyToManyField, AutoField
|
|
from django.contrib.gis.db.models import GeometryField
|
|
from assets.models import AbstractAsset, AbstractGeoAsset
|
|
|
|
excluded_types = (ForeignKey, OneToOneField, ManyToManyField, AutoField, GeometryField)
|
|
base_field_names = {f.name for f in AbstractAsset._meta.fields}
|
|
base_field_names |= {f.name for f in AbstractGeoAsset._meta.fields}
|
|
base_field_names.add('geojson')
|
|
|
|
for field in type(asset)._meta.fields:
|
|
if field.name in base_field_names:
|
|
continue
|
|
if isinstance(field, excluded_types):
|
|
continue
|
|
allowed_fields.add(field.name)
|
|
|
|
return list(allowed_fields)
|
|
|
|
|
|
def get_visible_fields_for_asset(user, asset):
|
|
"""
|
|
Get the list of fields visible to the user for a specific asset.
|
|
|
|
Priority (mirrors get_allowed_update_fields_for_asset):
|
|
1. UserAssetAccess.visible_fields (non-empty) → return it.
|
|
2. UserAssetTypeAccess.visible_fields (non-empty) → return it.
|
|
3. No restriction defined → return None (all fields visible).
|
|
"""
|
|
if not can_view_asset(user, asset):
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return None
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
ct = ContentType.objects.get_for_model(asset)
|
|
|
|
# Check per-instance override
|
|
try:
|
|
access = UserAssetAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
object_id=asset.pk,
|
|
)
|
|
if access.visible_fields:
|
|
return list(access.visible_fields)
|
|
except UserAssetAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Check per-type override
|
|
try:
|
|
type_access = UserAssetTypeAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
)
|
|
if type_access.visible_fields:
|
|
return list(type_access.visible_fields)
|
|
except UserAssetTypeAccess.DoesNotExist:
|
|
pass
|
|
|
|
# No restriction → all fields visible
|
|
return None
|
|
|
|
|
|
def get_allowed_actions_for_asset(user, asset):
|
|
"""
|
|
Get the list of actions the user is allowed to perform on assets.
|
|
"""
|
|
if not can_edit_asset(user, asset):
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
allowed_actions = set()
|
|
for role in user_config.roles.all():
|
|
role_actions = ASSET_ACTIONS_BY_ROLE.get(role.name, [])
|
|
allowed_actions.update(role_actions)
|
|
|
|
return list(allowed_actions)
|
|
|
|
|
|
def get_allowed_actions_for_thematic(user, thematic):
|
|
"""
|
|
Get the list of actions the user is allowed to perform for a thematic.
|
|
"""
|
|
can_view, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
if not can_edit:
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
allowed_actions = set()
|
|
for role in user_config.roles.all():
|
|
role_actions = ASSET_ACTIONS_BY_ROLE.get(role.name, [])
|
|
allowed_actions.update(role_actions)
|
|
|
|
return list(allowed_actions)
|
|
|
|
|
|
def asset_edit_permission_required(view_func):
|
|
"""
|
|
Decorator that checks if user has permission to edit assets.
|
|
Expects asset_id and asset_model as URL parameters.
|
|
"""
|
|
@wraps(view_func)
|
|
def wrapper(request, *args, **kwargs):
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
asset_model = kwargs.get('asset_model')
|
|
asset_id = kwargs.get('asset_id')
|
|
|
|
if not asset_model or not asset_id:
|
|
return HttpResponseForbidden(_("Missing asset information"))
|
|
|
|
try:
|
|
content_type = ContentType.objects.get(model=asset_model.lower())
|
|
model_class = content_type.model_class()
|
|
asset = get_object_or_404(model_class, pk=asset_id)
|
|
except ContentType.DoesNotExist:
|
|
return HttpResponseForbidden(_("Invalid asset type"))
|
|
|
|
if not can_edit_asset(request.user, asset):
|
|
return HttpResponseForbidden(_("You do not have permission to edit this asset"))
|
|
|
|
return view_func(request, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def can_edit_location_for_thematic(user, thematic_code):
|
|
"""
|
|
Check if the user has permission to create/edit/delete locations for a given thematic.
|
|
Relies on UserThematics.can_edit_locations.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
from common.models import Thematic
|
|
thematic = Thematic.objects.get(code=thematic_code)
|
|
user_thematic = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return user_thematic.can_edit_locations
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def location_asset_edit_permission_required(thematic_code):
|
|
"""
|
|
Decorator that checks if user has permission to edit assets for a thematic.
|
|
Used for location-based asset operations.
|
|
"""
|
|
def decorator(view_func):
|
|
@wraps(view_func)
|
|
def wrapper(request, *args, **kwargs):
|
|
from common.models import Thematic
|
|
|
|
try:
|
|
thematic = Thematic.objects.get(code=thematic_code)
|
|
except Thematic.DoesNotExist:
|
|
return HttpResponseForbidden(_("Invalid thematic"))
|
|
|
|
can_view, can_edit = get_user_asset_thematic_access(request.user, thematic)
|
|
if not can_edit:
|
|
return HttpResponseForbidden(_("You do not have permission to edit assets for this thematic"))
|
|
|
|
return view_func(request, *args, **kwargs)
|
|
|
|
return wrapper
|
|
return decorator
|