document-management/backend/dms/services/authorization_service.py

"""主案、子方案及共享附件的集中读取授权。"""

from __future__ import annotations

import logging
from dataclasses import dataclass

from sqlalchemy import func, select

from dms.common.enums import (
    AllowedAction,
    DocumentStatus,
    DocumentType,
    EnabledStatus,
    RoleCode,
    SECURITY_LEVEL_VALUES,
    SubjectType,
    VisibilityType,
)
from dms.extensions import db
from dms.models import AttachmentBinding, Document, Organization, Permission
from dms.security.auth_context import AuthContext
from dms.common.errors import SecurityLevelForbiddenError

logger = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class PlanAccess:
    allowed: bool
    reason: str | None
    source: Document | None
    permissions: tuple[Permission, ...]
    can_download: bool


def _security_allowed(document: Document, context: AuthContext) -> bool:
    return (
        SECURITY_LEVEL_VALUES[context.security_level]
        >= SECURITY_LEVEL_VALUES[type(context.security_level)(document.security_level)]
    )


def require_security_clearance(
    document: Document,
    context: AuthContext,
) -> None:
    """集中执行文档密级前置检查，ADMIN也不能绕过。"""
    if not _security_allowed(document, context):
        raise SecurityLevelForbiddenError()


def _active_organization_ancestor_ids(
    organization_id: int | None,
) -> set[int]:
    if organization_id is None:
        return set()
    organizations = db.session.scalars(
        select(Organization).where(
            Organization.is_deleted.is_(False),
            Organization.status == EnabledStatus.ENABLED.value,
        )
    ).all()
    by_id = {organization.id: organization for organization in organizations}
    result: set[int] = set()
    current_id: int | None = organization_id
    while current_id is not None and current_id not in result:
        current = by_id.get(current_id)
        if current is None:
            return set()
        result.add(current.id)
        current_id = current.parent_id
    return result


def _permission_source(document: Document) -> Document | None:
    if document.document_type == DocumentType.MAIN.value:
        return document
    if document.document_type != DocumentType.SUB_PLAN.value:
        return None
    root = db.session.scalar(
        select(Document).where(
            Document.id == document.root_document_id,
            Document.document_type == DocumentType.MAIN.value,
            Document.is_deleted.is_(False),
        )
    )
    if root is None:
        logger.error(
            "子方案数据一致性异常：document_id=%s root_document_id=%s",
            document.id,
            document.root_document_id,
        )
    return root


def evaluate_plan_access(
    document: Document,
    context: AuthContext,
) -> PlanAccess:
    """按固定顺序判断方案读取权限并返回有效权限快照。"""
    source = _permission_source(document)
    if source is None:
        return PlanAccess(False, "INVALID_ROOT", None, (), False)
    if (
        context.role_code == RoleCode.USER
        and document.document_status != DocumentStatus.PUBLISHED.value
    ):
        return PlanAccess(False, "STATUS", source, (), False)
    if not _security_allowed(document, context):
        return PlanAccess(False, "SECURITY", source, (), False)
    if context.role_code == RoleCode.AUDITOR:
        return PlanAccess(False, "ROLE", source, (), False)
    if context.role_code == RoleCode.ADMIN:
        return PlanAccess(True, None, source, (), True)

    visibility = VisibilityType(source.visibility_type)
    if visibility == VisibilityType.ALL_AUTHENTICATED:
        return PlanAccess(True, None, source, (), True)

    permissions = tuple(
        db.session.scalars(
            select(Permission).where(
                Permission.document_id == source.id,
                Permission.is_deleted.is_(False),
            )
        ).all()
    )
    ancestor_ids = _active_organization_ancestor_ids(context.organization_id)
    matched: list[Permission] = []
    for permission in permissions:
        if (
            permission.subject_type == SubjectType.ORG.value
            and permission.subject_id in ancestor_ids
        ):
            matched.append(permission)
        elif (
            visibility == VisibilityType.CUSTOM
            and permission.subject_type == SubjectType.USER.value
            and permission.subject_id == context.user_id
        ):
            matched.append(permission)
    if visibility == VisibilityType.ORGANIZATION:
        matched = [
            permission
            for permission in matched
            if permission.subject_type == SubjectType.ORG.value
        ]
    can_view = any(permission.can_view for permission in matched)
    return PlanAccess(
        can_view,
        None if can_view else "ACL",
        source,
        tuple(matched),
        can_view and any(permission.can_download for permission in matched),
    )


def plan_allowed_actions(
    document: Document,
    context: AuthContext,
    access: PlanAccess,
) -> list[str]:
    if not access.allowed:
        return []
    if context.role_code == RoleCode.USER:
        actions = [AllowedAction.VIEW.value]
        if access.can_download:
            actions.append(AllowedAction.DOWNLOAD.value)
        return actions
    if context.role_code != RoleCode.ADMIN:
        return []
    if document.document_type == DocumentType.SUB_PLAN.value:
        return [
            AllowedAction.VIEW.value,
            AllowedAction.DOWNLOAD.value,
            AllowedAction.EDIT.value,
            AllowedAction.DELETE.value,
        ]
    actions = [
        AllowedAction.VIEW.value,
        AllowedAction.DOWNLOAD.value,
        AllowedAction.EDIT.value,
        AllowedAction.CONFIG_PERMISSION.value,
        AllowedAction.DELETE.value,
        AllowedAction.BIND_ATTACHMENT.value,
    ]
    binding_count = db.session.scalar(
        select(func.count(AttachmentBinding.id)).where(
            AttachmentBinding.main_document_id == document.id,
            AttachmentBinding.is_deleted.is_(False),
        )
    )
    if binding_count:
        actions.append(AllowedAction.UNBIND_ATTACHMENT.value)
    return actions


def attachment_allowed_actions(context: AuthContext) -> list[str]:
    actions = [AllowedAction.VIEW.value, AllowedAction.DOWNLOAD.value]
    if context.role_code == RoleCode.ADMIN:
        actions.extend([AllowedAction.EDIT.value, AllowedAction.DELETE.value])
    return actions