Feature/add user business access clean

core/access.py

@@ -0,0 +1,302 @@
+import logging
+from datetime import datetime
+
+from django.contrib.auth.models import AnonymousUser
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import Q
+from django.http import JsonResponse
+
+from core.utils import (
+    get_business_access_cache,
+    get_content_type_cache,
+    get_current_user,
+    is_authentication_checked,
+    set_authentication_checked,
+)
+from core.models.user_business_access import UserBusinessAccess
+
+logger = logging.getLogger(__name__)
+
+SERVICE_AUTH_ERROR = {
+    "success": False,
+    "message": "Authentication required",
+    "detail": "PermissionDenied",
+}
+
+SERVICE_PERMISSION_ERROR = {
+    "success": False,
+    "message": "Permissions required",
+    "detail": "PermissionDenied",
+}
+
+# Shared format for business access checks:
+#   [content_type_label, object_id]
+#   [content_type_label, object_id, [perm_ids...]]
+ACCESS_REQUIREMENT_FORMAT = (
+    "List of [content_type_label, object_id] or "
+    "[content_type_label, object_id, [perm_ids...]] entries, "
+    "where content_type_label is 'app_label.modelname'."
+)
+
+
+def is_core_user(obj):
+    """Detect openIMIS core.User without importing User (avoids circular imports)."""
+    if obj is None or isinstance(obj, AnonymousUser):
+        return False
+
+    meta = getattr(obj, "_meta", None)
+    if meta is not None:
+        return getattr(meta, "label_lower", None) == "core.user"
+
+    cls = type(obj)
+    return (
+        cls.__name__ == "User"
+        and cls.__module__.endswith(".models.user")
+        and hasattr(obj, "has_perms")
+    )
+
+
+def is_authenticated_user(user):
+    return is_core_user(user) and getattr(user, "id", None)
+
+
+def is_http_request(obj):
+    return hasattr(obj, "META") and hasattr(obj, "method")
+
+
+def resolve_user(user_param, args, kwargs):
+    if is_core_user(user_param):
+        return user_param
+
+    resolved_user = get_current_user()
+    if resolved_user:
+        return resolved_user
+    if callable(user_param):
+        resolved = user_param(*args, **kwargs)
+        if resolved:
+            return resolved
+    elif user_param is not None and isinstance(user_param, str) and args:
+        resolved = getattr(args[0], user_param, None)
+        if resolved:
+            return resolved
+
+    if args and hasattr(args[0], "user") and not is_http_request(args[0]):
+        instance = args[0]
+        if instance.user:
+            return instance.user
+
+    return None
+
+
+def parse_access_requirement(requirement):
+    if not requirement or len(requirement) < 2:
+        return None, None, None
+
+    content_type_label = requirement[0]
+    object_id = requirement[1]
+    perms = requirement[2] if len(requirement) > 2 else None
+    return content_type_label, object_id, perms
+
+
+def _content_type_cache_key(content_type_label=None, *, app_label=None, business_object=None):
+    if business_object is not None:
+        return ("model", business_object.__class__.__module__, business_object.__class__.__name__)
+    if isinstance(content_type_label, ContentType):
+        return ("pk", content_type_label.pk)
+    if content_type_label is None:
+        return None
+
+    label = str(content_type_label)
+    if app_label:
+        return ("label", app_label, label.rsplit(".", 1)[-1].lower())
+    if "." in label:
+        app_label, model_name = label.rsplit(".", 1)
+        return ("label", app_label, model_name.lower())
+    return ("model_name", label.lower())
+
+
+def _business_access_cache_key(user, content_type, object_id):
+    user_id = getattr(user, "pk", None) or getattr(user, "id", None)
+    return (user_id, content_type.pk, str(object_id))
+
+
+def resolve_content_type(content_type_label=None, *, app_label=None, business_object=None):
+    if business_object is not None:
+        cache_key = _content_type_cache_key(business_object=business_object)
+    elif isinstance(content_type_label, ContentType):
+        return content_type_label
+    elif content_type_label is None:
+        return None
+    else:
+        cache_key = _content_type_cache_key(content_type_label, app_label=app_label)
+
+    if cache_key is not None:
+        content_type_cache = get_content_type_cache()
+        if cache_key in content_type_cache:
+            return content_type_cache[cache_key]
+
+    if business_object is not None:
+        content_type = ContentType.objects.get_for_model(business_object.__class__)
+    elif isinstance(content_type_label, ContentType):
+        content_type = content_type_label
+    else:
+        label = str(content_type_label)
+        if app_label:
+            model_name = label.rsplit(".", 1)[-1]
+            content_type = ContentType.objects.filter(
+                app_label=app_label,
+                model=model_name.lower(),
+            ).first()
+        elif "." in label:
+            app_label, model_name = label.rsplit(".", 1)
+            content_type = ContentType.objects.filter(
+                app_label=app_label,
+                model=model_name.lower(),
+            ).first()
+        else:
+            content_type = ContentType.objects.filter(model__iexact=label.lower()).first()
+
+    if cache_key is not None and content_type is not None:
+        content_type_cache[cache_key] = content_type
+    return content_type
+
+
+def has_business_access(user, *, content_type=None, object_id=None, business_object=None, now=None):
+    if not user:
+        return False
+
+    if business_object is not None:
+        content_type = resolve_content_type(business_object=business_object)
+        object_id = str(business_object.pk)
+
+    if not content_type or object_id is None:
+        return False
+
+    cache_key = _business_access_cache_key(user, content_type, object_id)
+    business_access_cache = get_business_access_cache()
+    if cache_key in business_access_cache:
+        return business_access_cache[cache_key]
+
+    now = now or datetime.now()
+    has_access = UserBusinessAccess.objects.filter(
+        user=user,
+        content_type=content_type,
+        object_id=str(object_id),
+        active=True,
+        date_valid_from__lte=now,
+    ).filter(
+        Q(date_valid_to__isnull=True) | Q(date_valid_to__gte=now)
+    ).exists()
+    business_access_cache[cache_key] = has_access
+    return has_access
+
+
+def has_role_perms(user, perm_list, *, list_evaluation_or=True):
+    if (
+        getattr(user, "is_superuser", False)
+        or getattr(user, "is_imis_admin", False)
+        or not perm_list
+    ):
+        return True
+    if list_evaluation_or:
+        return any(user.has_perm(perm) for perm in perm_list)
+    return all(user.has_perm(perm) for perm in perm_list)
+
+
+def satisfies_access_requirement(user, requirement, *, now=None):
+    content_type_label, object_id, perms = parse_access_requirement(requirement)
+    if not content_type_label:
+        return False
+
+    if perms and not has_role_perms(user, perms):
+        return False
+
+    app_label, model_name = content_type_label.rsplit(".", 1)
+    content_type = resolve_content_type(model_name, app_label=app_label)
+    if not content_type:
+        logger.error("Invalid content type: %s", content_type_label)
+        return False
+
+    return has_business_access(
+        user,
+        content_type=content_type,
+        object_id=object_id,
+        now=now,
+    )
+
+
+def evaluate_access_requirements(user, access_requirements, *, match_all=False, now=None):
+    if not access_requirements:
+        return True
+
+    now = now or datetime.now()
+    results = [
+        satisfies_access_requirement(user, requirement, now=now)
+        for requirement in access_requirements
+    ]
+    return all(results) if match_all else any(results)
+
+
+def user_has_permissions(
+    user,
+    permissions,
+    *,
+    access_requirements=None,
+    list_evaluation_or=True,
+):
+    if not is_authenticated_user(user):
+        return False
+    return user.has_perms(
+        permissions,
+        access_requirements=access_requirements,
+        list_evaluation_or=list_evaluation_or,
+    )
+
+
+def authentication_error(for_view=False):
+    if for_view:
+        return JsonResponse({"error": "Authentication required"}, status=401)
+    return SERVICE_AUTH_ERROR
+
+
+def permission_error(for_view=False):
+    if for_view:
+        return JsonResponse({"error": "Forbidden"}, status=403)
+    return SERVICE_PERMISSION_ERROR
+
+
+def guard_user_access(
+    *,
+    user_param,
+    args,
+    kwargs,
+    for_view=False,
+    require_auth=True,
+    permissions=None,
+    access_requirements=None,
+    list_evaluation_or=True,
+):
+    resolved_user = resolve_user(user_param, args, kwargs)
+
+    if require_auth and not is_authentication_checked():
+        if not is_authenticated_user(resolved_user):
+            return authentication_error(for_view)
+        set_authentication_checked()
+
+    if permissions is not None:
+        if not user_has_permissions(
+            resolved_user,
+            permissions,
+            access_requirements=access_requirements,
+            list_evaluation_or=list_evaluation_or,
+        ):
+            return permission_error(for_view)
+    elif access_requirements:
+        if not evaluate_access_requirements(
+            resolved_user,
+            access_requirements,
+            match_all=True,
+        ):
+            return permission_error(for_view)
+
+    return None

core/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
 from django.contrib.auth.models import Group, Permission
-from .models import FieldControl, ModuleConfiguration, TechnicalUser
+from .models import FieldControl, ModuleConfiguration, TechnicalUser, UserBusinessAccess
 from .forms import TechnicalUserAdmin, GroupAdmin
 
 admin.site.unregister(Group)
@@ -10,3 +10,11 @@
 admin.site.register(TechnicalUser, TechnicalUserAdmin)
 admin.site.register(Permission)
 admin.site.register(Group, GroupAdmin)
+
+
+@admin.register(UserBusinessAccess)
+class UserBusinessAccessAdmin(admin.ModelAdmin):
+    list_display = ['user', 'link_type', 'content_type', 'object_id', 'date_valid_from', 'date_valid_to', 'active']
+    list_filter = ['link_type', 'content_type', 'active']
+    search_fields = ['user__username', 'object_id']
+    raw_id_fields = ['user', 'content_type']

core/apps.py

@@ -248,7 +248,6 @@ def _configure_additional_settings(self, cfg):
 
     def ready(self):
         from .models import ModuleConfiguration
-
         cfg = ModuleConfiguration.get_or_default(MODULE_NAME, DEFAULT_CFG)
         self._configure_calendar(cfg)
         self._configure_user_config(cfg)