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,
    DeptScope,
    DocumentStatus,
    DocumentType,
    EnabledStatus,
    RoleCode,
    SECURITY_LEVEL_VALUES,
    SubjectType,
    VisibilityType,
)
from dms.extensions import db
from dms.models import (
    AttachmentBinding,
    Document,
    Organization,
    Permission,
    Role,
    RoleDepartment,
    RolePermission,
    User,
)
from dms.security.auth_context import AuthContext
from dms.common.errors import (
    DocumentDownloadForbiddenError,
    DocumentViewForbiddenError,
    SecurityLevelForbiddenError,
)

logger = logging.getLogger(__name__)


# 矩阵布尔位与 AllowedAction 的一一对应
MATRIX_ACTION_COLUMNS: tuple[tuple[str, AllowedAction], ...] = (
    ("can_view", AllowedAction.VIEW),
    ("can_download", AllowedAction.DOWNLOAD),
    ("can_edit", AllowedAction.EDIT),
    ("can_delete", AllowedAction.DELETE),
    ("can_config_permission", AllowedAction.CONFIG_PERMISSION),
    ("can_bind_attachment", AllowedAction.BIND_ATTACHMENT),
    ("can_unbind_attachment", AllowedAction.UNBIND_ATTACHMENT),
)


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


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 _effective_role(context: AuthContext) -> Role | None:
    """实时解析 USER 体系的生效角色：显式角色优先，未挂角色回退默认角色。

    显式角色被停用或软删时不回退默认角色（矩阵贡献为空）。
    """
    role_id = db.session.scalar(
        select(User.role_id).where(
            User.id == context.user_id,
            User.is_deleted.is_(False),
        )
    )
    if role_id is not None:
        return db.session.scalar(
            select(Role).where(
                Role.id == role_id,
                Role.is_deleted.is_(False),
                Role.status == EnabledStatus.ENABLED.value,
            )
        )
    return db.session.scalar(
        select(Role).where(
            Role.is_default.is_(True),
            Role.is_deleted.is_(False),
            Role.status == EnabledStatus.ENABLED.value,
        )
    )


def _dept_scope(
    source: Document, context: AuthContext, role: Role | None
) -> DeptScope:
    """按文档归属机构划分 OWN / ASSIGNED / OTHER。"""
    if source.organization_id is None:
        return DeptScope.OTHER
    if context.organization_id is not None:
        paths = db.session.execute(
            select(Organization.id, Organization.org_path).where(
                Organization.id.in_(
                    {source.organization_id, context.organization_id}
                ),
                Organization.is_deleted.is_(False),
            )
        ).all()
        by_id = {organization_id: path for organization_id, path in paths}
        doc_path = by_id.get(source.organization_id)
        user_path = by_id.get(context.organization_id)
        if doc_path and user_path:
            own_prefix = user_path if user_path.endswith("/") else user_path + "/"
            if doc_path == user_path or doc_path.startswith(own_prefix):
                return DeptScope.OWN
    if role is not None:
        assigned = db.session.scalar(
            select(RoleDepartment.id).where(
                RoleDepartment.role_id == role.id,
                RoleDepartment.organization_id == source.organization_id,
                RoleDepartment.is_deleted.is_(False),
            )
        )
        if assigned is not None:
            return DeptScope.ASSIGNED
    return DeptScope.OTHER


def _matrix_actions(source: Document, context: AuthContext) -> frozenset[str]:
    """取生效角色矩阵中命中的操作位集合（密级×部门范围，实时查库）。"""
    role = _effective_role(context)
    if role is None:
        return frozenset()
    scope = _dept_scope(source, context, role)
    row = db.session.scalar(
        select(RolePermission).where(
            RolePermission.role_id == role.id,
            RolePermission.security_level == source.security_level,
            RolePermission.dept_scope == scope.value,
            RolePermission.is_deleted.is_(False),
        )
    )
    if row is None:
        return frozenset()
    return frozenset(
        action.value
        for column, action in MATRIX_ACTION_COLUMNS
        if getattr(row, column)
    )


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.ADMIN:
        return PlanAccess(True, None, source, (), True)

    matrix_actions = _matrix_actions(source, context)

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

    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)
    # 矩阵与单文档 ACL 为 OR 叠加
    can_view = can_view or AllowedAction.VIEW.value in matrix_actions
    can_download = (can_view and any(
        permission.can_download for permission in matched
    )) or AllowedAction.DOWNLOAD.value in matrix_actions
    return PlanAccess(
        can_view,
        None if can_view else "ACL",
        source,
        tuple(matched),
        can_view and can_download,
        matrix_actions,
    )


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)
        # 矩阵操作位 OR 叠加；子方案不适用授权/挂载/解除挂载
        applicable = None
        if document.document_type == DocumentType.SUB_PLAN.value:
            applicable = {
                AllowedAction.VIEW.value,
                AllowedAction.DOWNLOAD.value,
                AllowedAction.EDIT.value,
                AllowedAction.DELETE.value,
            }
        for action in access.matrix_actions:
            if applicable is not None and action not in applicable:
                continue
            if action not in actions:
                actions.append(action)
        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,
    access: PlanAccess | None = None,
) -> list[str]:
    if context.role_code == RoleCode.ADMIN:
        return [
            AllowedAction.VIEW.value,
            AllowedAction.DOWNLOAD.value,
            AllowedAction.EDIT.value,
            AllowedAction.DELETE.value,
        ]
    actions: list[str] = []
    if access is not None and access.allowed:
        actions.append(AllowedAction.VIEW.value)
        if access.can_download:
            actions.append(AllowedAction.DOWNLOAD.value)
    return actions


def evaluate_attachment_access(
    attachment: Document,
    context: AuthContext,
) -> tuple[bool, bool]:
    """USER只能通过至少一个有权有效主案访问共享附件。"""
    if context.role_code == RoleCode.ADMIN:
        return True, True
    mains = db.session.scalars(
        select(Document)
        .join(
            AttachmentBinding,
            AttachmentBinding.main_document_id == Document.id,
        )
        .where(
            AttachmentBinding.attachment_document_id == attachment.id,
            AttachmentBinding.is_deleted.is_(False),
            Document.document_type == DocumentType.MAIN.value,
            Document.is_deleted.is_(False),
        )
        .order_by(Document.id.asc())
    ).all()
    can_view = False
    can_download = False
    for main in mains:
        access = evaluate_plan_access(main, context)
        can_view = can_view or access.allowed
        can_download = can_download or (access.allowed and access.can_download)
    return can_view, can_download


def require_attachment_access(
    attachment: Document,
    context: AuthContext,
    *,
    download: bool,
) -> tuple[bool, bool]:
    can_view, can_download = evaluate_attachment_access(attachment, context)
    if download and not can_download:
        raise DocumentDownloadForbiddenError()
    if not download and not can_view:
        raise DocumentViewForbiddenError()
    return can_view, can_download