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

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

from __future__ import annotations

import logging
from dataclasses import dataclass

from sqlalchemy import and_, exists, false, func, or_, select, true

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,
    RolePermission,
    User,
)
from dms.security.auth_context import AuthContext
from dms.common.errors import (
    DocumentActionForbiddenError,
    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()


# 写路径操作码 → 中文动词（拒绝提示与前端口径一致）
ACTION_LABELS: dict[str, str] = {
    AllowedAction.VIEW.value: "查看",
    AllowedAction.DOWNLOAD.value: "下载",
    AllowedAction.EDIT.value: "编辑",
    AllowedAction.DELETE.value: "删除",
    AllowedAction.CONFIG_PERMISSION.value: "授权",
    AllowedAction.BIND_ATTACHMENT.value: "挂载",
    AllowedAction.UNBIND_ATTACHMENT.value: "解除挂载",
}


def require_plan_action(
    document: Document,
    context: AuthContext,
    action: AllowedAction,
) -> PlanAccess | None:
    """写路径集中鉴权：密级前置 + 文档访问 + 操作位校验。

    与读路径/前端按钮同源（evaluate_plan_access + plan_allowed_actions），
    保证"按钮可点 ⟺ 后端放行"：

    - 密级不足：任何角色（含 ADMIN）一律拒绝；
    - 共享附件：矩阵不授予编辑/删除/授权/挂载位，仅 ADMIN 可维护；
    - 主案/子方案：ADMIN 恒过（密级通过后）；USER 由矩阵操作位判定，
      子方案自动过滤授权/挂载/解除挂载位（继承自主案）。
    """
    if not _security_allowed(document, context):
        raise SecurityLevelForbiddenError()
    label = ACTION_LABELS.get(action.value, "执行该操作")
    if document.document_type == DocumentType.ATTACHMENT.value:
        if context.role_code != RoleCode.ADMIN:
            raise DocumentActionForbiddenError(
                f"无权{label}该共享附件：仅管理员可维护共享附件"
            )
        return None
    access = evaluate_plan_access(document, context)
    if not access.allowed:
        raise DocumentActionForbiddenError(f"无权{label}该文档")
    actions = plan_allowed_actions(document, context, access)
    if action.value not in actions:
        raise DocumentActionForbiddenError(
            f"无权{label}该文档：角色权限未授予该操作"
        )
    return access


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) -> DeptScope:
    """按文档归属机构划分 OWN / OTHER（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
    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)
    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,
        ]
    # 阶段11方案B：ADMIN恒含UNBIND（与BIND对称）。原"存在附件挂载才有位"
    # 是UI显影残留——附件表零挂载不渲染、子案挂载(0018)复用该位后反而出现
    # "仅有子案挂载的主案管理员无法解挂"的倒挂，故移除条件（顺带消除
    # 主案列表每行一次的count查询）。
    return [
        AllowedAction.VIEW.value,
        AllowedAction.DOWNLOAD.value,
        AllowedAction.EDIT.value,
        AllowedAction.CONFIG_PERMISSION.value,
        AllowedAction.DELETE.value,
        AllowedAction.BIND_ATTACHMENT.value,
        AllowedAction.UNBIND_ATTACHMENT.value,
    ]


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只能通过至少一个有权有效主案访问共享附件。

    阶段11：附件允许携带任意密级（0020放宽shape约束，存量子案保留原密级），
    读路径密级前置与主案/子方案同口径——任何角色（含ADMIN）密级不足即拒绝。
    """
    if not _security_allowed(attachment, context):
        return False, False
    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]:
    # 密级不足优先按密级拒绝（前端据此提示"密级不足"而非"权限不足"）
    if not _security_allowed(attachment, context):
        raise SecurityLevelForbiddenError()
    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


# 管理类操作位：命中任一即视为“对该文档有管理能力”，用于动态展示后台管理菜单。
MANAGE_MATRIX_COLUMNS: tuple[str, ...] = (
    "can_edit",
    "can_delete",
    "can_config_permission",
    "can_bind_attachment",
    "can_unbind_attachment",
)


def _organization_subtree_ids(organization_id: int | None) -> set[int]:
    """用户机构的下属子树（含自身），与 _dept_scope 的 OWN 判定同口径。"""
    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}
    root = by_id.get(organization_id)
    if root is None:
        return set()
    prefix = root.org_path if root.org_path.endswith("/") else root.org_path + "/"
    return {
        organization.id
        for organization in organizations
        if organization.id == organization_id
        or (organization.org_path or "").startswith(prefix)
    }


def has_any_manage_action(context: AuthContext) -> bool:
    """用户是否对任一已发布主案具备管理类操作（编辑/删除/授权/挂载/解除挂载）。

    与 plan_allowed_actions 同源：ADMIN 恒真；USER 由角色权限矩阵推导——
    矩阵行命中（密级×部门范围）且存在“可查看”的对应主案时管理位才生效。
    只看主案即可完整覆盖子案：子案矩阵取自其根主案的同一行，主案命中
    子案必命中。用于登录/会话接口动态计算 allowedModules 中的
    BACKEND_MANAGEMENT，替代“管理员固定可见”的角色硬编码。
    """
    if context.role_code == RoleCode.ADMIN:
        return True
    role = _effective_role(context)
    if role is None:
        return False
    rows = db.session.scalars(
        select(RolePermission).where(
            RolePermission.role_id == role.id,
            RolePermission.is_deleted.is_(False),
        )
    ).all()
    candidates = [
        row
        for row in rows
        if any(getattr(row, column) for column in MANAGE_MATRIX_COLUMNS)
    ]
    if not candidates:
        return False

    # 部门范围两区集合：OWN 子树 / 其余（含无归属）。
    own_ids = _organization_subtree_ids(context.organization_id)
    ancestor_ids = _active_organization_ancestor_ids(context.organization_id)
    # ORG 主体授权对机构/自定义可见均生效；USER 主体仅自定义可见生效。
    org_subject = and_(
        Permission.subject_type == SubjectType.ORG.value,
        (
            Permission.subject_id.in_(ancestor_ids)
            if ancestor_ids
            else false()
        ),
    )
    user_subject = and_(
        Permission.subject_type == SubjectType.USER.value,
        Permission.subject_id == context.user_id,
        Document.visibility_type == VisibilityType.CUSTOM.value,
    )
    acl_view_granted = exists(
        select(Permission.id).where(
            Permission.document_id == Document.id,
            Permission.is_deleted.is_(False),
            Permission.can_view.is_(True),
            or_(org_subject, user_subject),
        )
    )

    for row in candidates:
        scope = DeptScope(row.dept_scope)
        if scope == DeptScope.OWN:
            scope_filter = Document.organization_id.in_(own_ids)
        else:
            scope_filter = or_(
                Document.organization_id.is_(None),
                Document.organization_id.not_in(own_ids),
            )
        # 查看权三来源（与 evaluate_plan_access 同口径）：
        # ① 本矩阵行勾选查看；② 全员可见；③ 文档级 ACL 授权查看。
        view_sources: list = [
            Document.visibility_type == VisibilityType.ALL_AUTHENTICATED.value,
            acl_view_granted,
        ]
        if row.can_view:
            view_sources.append(true())
        hit = db.session.scalar(
            select(Document.id).where(
                Document.is_deleted.is_(False),
                Document.document_status == DocumentStatus.PUBLISHED.value,
                Document.document_type == DocumentType.MAIN.value,
                Document.security_level == row.security_level,
                scope_filter,
                or_(*view_sources),
            ).limit(1)
        )
        if hit is not None:
            return True
    return False