258 lines
9.4 KiB
Python
258 lines
9.4 KiB
Python
# backend/api/v1/documents/permissions.py
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
from django.db.models import Q, QuerySet
|
|
from rest_framework.permissions import (
|
|
SAFE_METHODS,
|
|
BasePermission,
|
|
DjangoModelPermissions,
|
|
)
|
|
from rest_framework.request import Request
|
|
from rest_framework.views import View
|
|
|
|
from ..permissions import AdminOrActionOrModelPerms
|
|
|
|
|
|
def _is_admin(user) -> bool:
|
|
return bool(
|
|
user
|
|
and user.is_authenticated
|
|
and (getattr(user, "is_staff", False) or getattr(user, "is_superuser", False))
|
|
)
|
|
|
|
|
|
def _request_sets_regulation(request: Request) -> bool:
|
|
raw_value = request.data.get("is_regulation")
|
|
if isinstance(raw_value, bool):
|
|
return raw_value
|
|
return str(raw_value or "").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _has_any_model_perm(
|
|
user,
|
|
app_label: str,
|
|
model_name: str,
|
|
actions=("view", "change", "add", "delete"),
|
|
) -> bool:
|
|
"""Проверяет наличие любого модельного права из набора."""
|
|
for act in actions:
|
|
code = f"{act}_{model_name}"
|
|
if user.has_perm(f"{app_label}.{code}"):
|
|
return True
|
|
return False
|
|
|
|
|
|
def user_has_any_document_model_perm(user) -> bool:
|
|
return _has_any_model_perm(user, "documents", "document")
|
|
|
|
|
|
def _user_has_document_access(user, obj) -> bool:
|
|
"""Проверяет пользовательский доступ к конкретному документу."""
|
|
if getattr(obj, "sent_to_all", False) and getattr(user, "is_active", False):
|
|
return True
|
|
|
|
if getattr(obj, "uploaded_by_id", None) == getattr(user, "pk", None):
|
|
return True
|
|
|
|
if obj.recipients.filter(pk=user.pk, is_active=True).exists():
|
|
return True
|
|
|
|
if obj.departments.filter(
|
|
employeedepartment__employee=user,
|
|
employeedepartment__is_active=True,
|
|
).exists():
|
|
return True
|
|
|
|
return obj.departments.filter(
|
|
roles__assignments__employee=user,
|
|
roles__assignments__is_active=True,
|
|
).exists()
|
|
|
|
|
|
def filter_documents_for_user(queryset: QuerySet, user) -> QuerySet:
|
|
"""Ограничить queryset документами, которые пользователь вправе читать."""
|
|
if not (user and user.is_authenticated):
|
|
return queryset.none()
|
|
if user.is_staff or user.is_superuser or user.has_perm("documents.view_document"):
|
|
return queryset
|
|
|
|
return filter_documents_for_audience(queryset, user)
|
|
|
|
|
|
def filter_documents_for_audience(queryset: QuerySet, user) -> QuerySet:
|
|
"""Apply recipient/department/owner visibility even for privileged users."""
|
|
return queryset.filter(
|
|
Q(sent_to_all=True)
|
|
| Q(recipients=user)
|
|
| Q(
|
|
departments__employeedepartment__employee=user,
|
|
departments__employeedepartment__is_active=True,
|
|
)
|
|
| Q(
|
|
departments__roles__assignments__employee=user,
|
|
departments__roles__assignments__is_active=True,
|
|
)
|
|
| Q(uploaded_by=user)
|
|
).distinct()
|
|
|
|
|
|
def user_can_view_document(user, document) -> bool:
|
|
"""Проверить доступ к документу без зависимости от DRF view/action."""
|
|
if not (user and user.is_authenticated):
|
|
return False
|
|
if user.is_staff or user.is_superuser or user.has_perm("documents.view_document"):
|
|
return True
|
|
return _user_has_document_access(user, document)
|
|
|
|
|
|
class FolderPermission(BasePermission):
|
|
"""Папки видны сотрудникам, но менять их можно только владельцу/менеджеру."""
|
|
|
|
def has_permission(self, request: Request, view: View) -> bool:
|
|
return bool(request.user and request.user.is_authenticated)
|
|
|
|
def has_object_permission(self, request: Request, view: View, obj: Any) -> bool:
|
|
user = request.user
|
|
if request.method in SAFE_METHODS:
|
|
return True
|
|
if user.is_staff or user.is_superuser:
|
|
return True
|
|
if request.method == "DELETE":
|
|
return getattr(obj, "owner_id", None) == user.pk or user.has_perm(
|
|
"documents.delete_document"
|
|
)
|
|
return getattr(obj, "owner_id", None) == user.pk or user.has_perm(
|
|
"documents.change_document"
|
|
)
|
|
|
|
|
|
class DocumentTagPermission(BasePermission):
|
|
"""Чтение/создание тегов доступно сотрудникам, изменение — менеджерам."""
|
|
|
|
def has_permission(self, request: Request, view: View) -> bool:
|
|
user = request.user
|
|
if not (user and user.is_authenticated):
|
|
return False
|
|
if request.method in SAFE_METHODS or getattr(view, "action", None) == "create":
|
|
return True
|
|
return (
|
|
user.is_staff
|
|
or user.is_superuser
|
|
or user.has_perm("documents.change_document")
|
|
)
|
|
|
|
def has_object_permission(self, request: Request, view: View, obj: Any) -> bool:
|
|
return self.has_permission(request, view)
|
|
|
|
|
|
class DocumentReadOrModelPerms(AdminOrActionOrModelPerms):
|
|
"""Чтение/скачивание/ознакомление: любой аутентифицированный,
|
|
но доступ только к 'своим' документам (или если есть модельные права/админ).
|
|
Создание/редактирование/удаление: только админы и обладатели модельных прав.
|
|
|
|
Raises:
|
|
PermissionDenied: если нет доступа к конкретному документу.
|
|
"""
|
|
|
|
perms_map = deepcopy(DjangoModelPermissions.perms_map)
|
|
perms_map["GET"] = []
|
|
perms_map["HEAD"] = []
|
|
perms_map["OPTIONS"] = []
|
|
|
|
def has_permission(self, request: Request, view: View) -> bool:
|
|
action = getattr(view, "action", None)
|
|
|
|
# SAFE + acknowledge/download → достаточно быть аутентифицированным
|
|
if request.method in SAFE_METHODS or action in (
|
|
"acknowledge",
|
|
"download",
|
|
"archive",
|
|
"comments",
|
|
"document_comment",
|
|
"set_maximally_hidden",
|
|
):
|
|
return bool(request.user and request.user.is_authenticated)
|
|
|
|
# Обычные документы могут создавать все аутентифицированные сотрудники,
|
|
# регламенты — только администраторы.
|
|
if action == "create":
|
|
if _request_sets_regulation(request):
|
|
return _is_admin(request.user)
|
|
return bool(request.user and request.user.is_authenticated)
|
|
|
|
# update/destroy и related documents - проверяем на уровне объекта
|
|
if action in (
|
|
"update",
|
|
"partial_update",
|
|
"destroy",
|
|
"add_related",
|
|
"remove_related",
|
|
"revert",
|
|
):
|
|
return bool(request.user and request.user.is_authenticated)
|
|
|
|
# небезопасные → staff ИЛИ стандартные модельные права
|
|
# (add/change/delete)
|
|
if request.user and request.user.is_staff:
|
|
return True
|
|
return super().has_permission(request, view)
|
|
|
|
@staticmethod
|
|
def _unprivileged_object_permission(request, action, obj) -> bool:
|
|
user = request.user
|
|
if action in {"revert", "add_related", "remove_related"}:
|
|
return user.has_perm("documents.change_document", obj)
|
|
|
|
if action in {"comments", "document_comment"}:
|
|
return _user_has_document_access(user, obj)
|
|
|
|
model_action_by_method = {
|
|
"PUT": "change",
|
|
"PATCH": "change",
|
|
"DELETE": "delete",
|
|
}
|
|
model_action = model_action_by_method.get(request.method)
|
|
if model_action:
|
|
return user.has_perm(
|
|
f"documents.{model_action}_document",
|
|
obj,
|
|
)
|
|
|
|
# SAFE/ack/download без модельных прав → только пользователи,
|
|
# которым документ доступен.
|
|
if request.method in SAFE_METHODS or action in {
|
|
"acknowledge",
|
|
"download",
|
|
"archive",
|
|
"comments",
|
|
"document_comment",
|
|
"set_maximally_hidden",
|
|
}:
|
|
return _user_has_document_access(user, obj)
|
|
|
|
# Небезопасные без прав — нельзя (POST custom actions)
|
|
return False
|
|
|
|
def has_object_permission(self, request: Request, view: View, obj: Any) -> bool:
|
|
user = request.user
|
|
if not (user and user.is_authenticated):
|
|
return False
|
|
|
|
action = getattr(view, "action", None)
|
|
regulation_update = action in {"update", "partial_update"} and (
|
|
getattr(obj, "is_regulation", False) or _request_sets_regulation(request)
|
|
)
|
|
regulation_action = getattr(obj, "is_regulation", False) and action in {
|
|
"add_related",
|
|
"remove_related",
|
|
"revert",
|
|
}
|
|
if regulation_update or regulation_action:
|
|
return _is_admin(user)
|
|
if user.is_staff or _has_any_model_perm(user, "documents", "document"):
|
|
return True
|
|
return self._unprivileged_object_permission(request, action, obj)
|