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

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

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import datetime, timezone

from sqlalchemy import exists, or_, select, true

from dms.common.enums import (
    AllowedAction,
    DeptScope,
    DocumentStatus,
    DocumentType,
    EnabledStatus,
    RoleCode,
    SECURITY_LEVEL_VALUES,
)
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,
)
from dms.services.admin_permission import SUPER_ADMIN_USERNAME

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()


class AccessMemo:
    """请求级授权备忘：单次列表批量评估共享，缓存请求内不变量
    （生效角色、机构路径、矩阵行、祖先机构、权限源主案、授权列表），
    消除逐行重复查库（对齐 Java 端 AccessMemo）。

    仅限同一请求、同一用户上下文内使用，禁止跨请求复用——
    备忘不感知请求期间的权限数据变更。
    """

    __slots__ = (
        "role_loaded",
        "role",
        "org_paths_by_id",
        "matrix_actions_by_key",
        "roots_by_id",
        "permissions_by_source_id",
    )

    def __init__(self) -> None:
        # 生效角色是否已加载（role 允许为 None：无生效角色）
        self.role_loaded = False
        self.role: Role | None = None
        # 机构 org_path 按 id 缓存（dept_scope 逐行判定复用）
        self.org_paths_by_id: dict[int, str | None] = {}
        # 矩阵操作位按 (security_level, dept_scope) 缓存
        self.matrix_actions_by_key: dict[tuple[str, str], frozenset[str]] = {}
        # SUB_PLAN 权限源主案按 root_id 缓存（同根子案批量评估复用）
        self.roots_by_id: dict[int, Document | None] = {}
        # 授权列表按权限源文档 id 缓存（evaluate_plan_access 与权限摘要共享）
        self.permissions_by_source_id: dict[int, list[Permission]] = {}


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 _is_super_admin(context: AuthContext) -> bool:
    """内置超管（username == admin）：文档操作不受角色矩阵限制。

    其余 ADMIN（受限管理员）文档权限与 USER 同口径，按关联角色的
    权限矩阵（密级×部门范围×操作位）判定。
    """
    return (
        context.role_code == RoleCode.ADMIN
        and context.username == SUPER_ADMIN_USERNAME
    )


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 与 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 _permission_source(
    document: Document,
    memo: AccessMemo | None = None,
) -> Document | None:
    if document.document_type == DocumentType.MAIN.value:
        return document
    if document.document_type != DocumentType.SUB_PLAN.value:
        return None
    if memo is not None and document.root_document_id in memo.roots_by_id:
        return memo.roots_by_id[document.root_document_id]
    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,
        )
    if memo is not None:
        memo.roots_by_id[document.root_document_id] = root
    return root


def _effective_role(
    context: AuthContext,
    memo: AccessMemo | None = None,
) -> Role | None:
    """实时解析 USER 体系的生效角色：显式角色优先，未挂角色回退默认角色。

    显式角色被停用或软删时不回退默认角色（矩阵贡献为空）。
    """
    if memo is not None and memo.role_loaded:
        return memo.role
    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:
        role = db.session.scalar(
            select(Role).where(
                Role.id == role_id,
                Role.is_deleted.is_(False),
                Role.status == EnabledStatus.ENABLED.value,
            )
        )
    else:
        role = db.session.scalar(
            select(Role).where(
                Role.is_default.is_(True),
                Role.is_deleted.is_(False),
                Role.status == EnabledStatus.ENABLED.value,
            )
        )
    if memo is not None:
        memo.role_loaded = True
        memo.role = role
    return role


def _org_path(organization_id: int, memo: AccessMemo | None) -> str | None:
    if memo is not None and organization_id in memo.org_paths_by_id:
        return memo.org_paths_by_id[organization_id]
    path = db.session.scalar(
        select(Organization.org_path).where(
            Organization.id == organization_id,
            Organization.is_deleted.is_(False),
        )
    )
    if memo is not None:
        memo.org_paths_by_id[organization_id] = path
    return path


def _dept_scope(
    source: Document,
    context: AuthContext,
    memo: AccessMemo | None = None,
) -> DeptScope:
    """按文档归属机构划分 OWN / OTHER（OTHER 涵盖所属部门子树外全部部门）。"""
    if source.organization_id is None:
        return DeptScope.OTHER
    if context.organization_id is not None:
        if memo is not None:
            doc_path = _org_path(source.organization_id, memo)
            user_path = _org_path(context.organization_id, memo)
        else:
            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,
    memo: AccessMemo | None = None,
) -> frozenset[str]:
    """取生效角色矩阵中命中的操作位集合（密级×部门范围，实时查库）。"""
    role = _effective_role(context, memo)
    if role is None:
        return frozenset()
    scope = _dept_scope(source, context, memo)
    cache_key = (source.security_level, scope.value)
    if memo is not None and cache_key in memo.matrix_actions_by_key:
        return memo.matrix_actions_by_key[cache_key]
    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:
        actions = frozenset()
    else:
        actions = frozenset(
            action.value
            for column, action in MATRIX_ACTION_COLUMNS
            if getattr(row, column)
        )
    if memo is not None:
        memo.matrix_actions_by_key[cache_key] = actions
    return actions


def _active_permissions(
    source_id: int,
    memo: AccessMemo | None = None,
) -> list[Permission]:
    """权限源文档的授权列表（evaluate_plan_access 与权限摘要共享缓存）。"""
    if memo is not None and source_id in memo.permissions_by_source_id:
        return memo.permissions_by_source_id[source_id]
    permissions = list(
        db.session.scalars(
            select(Permission).where(
                Permission.document_id == source_id,
                Permission.is_deleted.is_(False),
            )
        ).all()
    )
    if memo is not None:
        memo.permissions_by_source_id[source_id] = permissions
    return permissions


def evaluate_plan_access(
    document: Document,
    context: AuthContext,
    memo: AccessMemo | None = None,
) -> PlanAccess:
    """按固定顺序判断方案读取权限并返回有效权限快照。"""
    source = _permission_source(document, memo)
    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)
    # 超管直接放行；受限 ADMIN 落入下方矩阵+ACL 路径（与 USER 同口径，
    # 但不受发布状态限制——状态豁免仍是管理员语义）
    if _is_super_admin(context):
        return PlanAccess(True, None, source, (), True)

    matrix_actions = _matrix_actions(source, context, memo)

    # 借阅语义（0026）：ACL 仅面向用户且限时生效，到期即失效（惰性判定，
    # 无需定时任务）；矩阵与单文档借阅 ACL 为 OR 叠加
    now = datetime.now(timezone.utc).replace(tzinfo=None)
    matched = [
        permission
        for permission in _active_permissions(source.id, memo)
        if permission.subject_id == context.user_id
        and permission.expires_at > now
    ]
    can_view = any(permission.can_view for permission in matched)
    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 not _is_super_admin(context):
        # 受限 ADMIN 与 USER 同口径：基础查看/下载 + 矩阵操作位 OR 叠加；
        # 子方案不适用授权/挂载/解除挂载
        actions = [AllowedAction.VIEW.value]
        if access.can_download:
            actions.append(AllowedAction.DOWNLOAD.value)
        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 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
    memo = AccessMemo()
    for main in mains:
        access = evaluate_plan_access(main, context, memo)
        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（ADMIN 因管理位独立、
    审计等模块恒可见，不依赖本函数）。
    """
    if _is_super_admin(context):
        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)
    # 借阅语义（0026）：ACL 仅面向用户且限时生效（到期即失效）
    acl_view_granted = exists(
        select(Permission.id).where(
            Permission.document_id == Document.id,
            Permission.is_deleted.is_(False),
            Permission.can_view.is_(True),
            Permission.subject_id == context.user_id,
            Permission.expires_at
            > datetime.now(timezone.utc).replace(tzinfo=None),
        )
    )

    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 = [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