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

"""管理类权限判定：分类/机构/用户编辑开关与管理范围锚点。

语义（管理权限矩阵化后）：
- 三管理位（manage_category_edit / manage_org_edit / manage_user_edit）挂在
  角色上，判定读"生效角色"（用户显式角色，未挂角色回退默认角色），
  且仅对 ADMIN 类别生效；分类编辑全局生效，机构/用户编辑受管理范围约束；
- 角色上 managed_org_id 为管理范围锚点：NULL=不受范围约束（全局），否则
  机构/用户编辑限定锚点部门及其下级子树（沿启用边BFS）；
- 内置 admin 账号恒为全权超管，不受开关与范围约束。
"""

from __future__ import annotations

from sqlalchemy import select

from dms.common.enums import EnabledStatus, RoleCode
from dms.common.errors import ForbiddenError
from dms.models import Organization, Role, User
from dms.security.auth_context import get_auth_context

SUPER_ADMIN_USERNAME = "admin"

# managed_org_id 哨兵值 0 = "SELF"（锚点跟随用户所在部门）：
# 0 不是合法机构ID，机构/用户编辑范围按各用户自身 organization_id 收敛。
MANAGED_ORG_SELF_ANCHOR = 0
MANAGED_ORG_SELF = "SELF"


def effective_role_snapshot(session, role_id: int | None) -> Role | None:
    """按显式角色ID解析生效角色快照（活跃且启用优先，回退默认角色）。

    与 effective_manage_role 同语义，用于尚未落库的目标用户
    （创建/编辑用户时的防提权校验）。
    """
    if role_id is not None:
        role = session.get(Role, role_id)
        if (
            role is not None
            and not role.is_deleted
            and role.status == EnabledStatus.ENABLED.value
        ):
            return role
    return session.scalar(
        select(Role).where(
            Role.is_default.is_(True),
            Role.is_deleted.is_(False),
            Role.status == EnabledStatus.ENABLED.value,
        )
    )


def effective_manage_role(session, user: User) -> Role | None:
    """生效角色：显式角色（活跃且启用）优先，否则回退默认角色（活跃且启用）。"""
    return effective_role_snapshot(session, user.role_id)


def is_super_admin(user: User) -> bool:
    """内置 admin 账号恒为超管，不受三管理位与管理范围约束。"""
    return user.username == SUPER_ADMIN_USERNAME


def _has_manage_flag(session, user: User, flag: str) -> bool:
    if user.role_code != RoleCode.ADMIN.value:
        return False
    role = effective_manage_role(session, user)
    return role is not None and bool(getattr(role, flag))


def can_edit_categories(session, user: User) -> bool:
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_category_edit")


def can_edit_organizations(session, user: User) -> bool:
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_org_edit")


def can_edit_users(session, user: User) -> bool:
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_user_edit")


def can_manage_system(session, user: User) -> bool:
    """系统管理页签（契约1.23）：manage_system 管理位。"""
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_system")


def can_view_audit_log(session, user: User) -> bool:
    """日志审计页签（契约1.23）：manage_audit_log 管理位。"""
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_audit_log")


def can_edit_roles(session, user: User) -> bool:
    """角色管理写操作（契约1.25）：manage_role_edit 管理位。

    角色为全局对象，不受管理范围锚点约束（同分类编辑）。
    """
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_role_edit")


def require_category_editor(session) -> User:
    """取当前认证用户并校验分类编辑权限。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_categories(session, actor):
        raise ForbiddenError("无分类编辑权限")
    return actor


def require_organization_editor(session) -> User:
    """取当前认证用户并校验机构编辑权限（范围按目标另行判定）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_organizations(session, actor):
        raise ForbiddenError("无机构编辑权限")
    return actor


def require_user_editor(session) -> User:
    """取当前认证用户并校验用户编辑权限（范围按目标另行判定）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_users(session, actor):
        raise ForbiddenError("无用户编辑权限")
    return actor


def require_system_manager(session) -> User:
    """取当前认证用户并校验系统管理权限（契约1.23）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_manage_system(session, actor):
        raise ForbiddenError("无系统管理权限")
    return actor


def require_audit_log_viewer(session) -> User:
    """取当前认证用户并校验日志审计查看权限（契约1.23）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_view_audit_log(session, actor):
        raise ForbiddenError("无日志审计权限")
    return actor


def require_role_editor(session) -> User:
    """取当前认证用户并校验角色编辑权限（契约1.25）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_roles(session, actor):
        raise ForbiddenError("无角色编辑权限")
    return actor


def managed_scope_ids(session, user: User) -> set[int] | None:
    """操作者可管理的机构ID集合；None 表示不受范围约束（全局）。

    实际范围以生效角色锚点为准（契约1.30）：super_admin 与锚点 NULL 的
    管理员不受约束；角色锚点为 SELF 哨兵（0）时回落用户级管理范围
    （sys_user.managed_org_id）——NULL=全局、SELF（0）=用户所在机构
    （无部门则无可管理范围）、机构ID=固定锚点；其余按锚点部门子树
    （仅沿启用边）收敛。
    """
    if is_super_admin(user):
        return None
    role = effective_manage_role(session, user)
    anchor = role.managed_org_id if role is not None else None
    if anchor is None:
        return None
    if anchor == MANAGED_ORG_SELF_ANCHOR:
        # 角色「跟随用户设置」：取用户自身管理范围（契约1.30）
        anchor = user.managed_org_id
        if anchor is None:
            return None
        if anchor == MANAGED_ORG_SELF_ANCHOR:
            # 用户范围「跟随所在机构」：按用户所在部门收敛
            if user.organization_id is None:
                return set()
            anchor = user.organization_id
    organizations = session.scalars(
        select(Organization).where(Organization.is_deleted.is_(False))
    ).all()
    children: dict[int | None, list[int]] = {}
    for item in organizations:
        if item.status == EnabledStatus.ENABLED.value:
            children.setdefault(item.parent_id, []).append(item.id)
    scope = {anchor}
    pending = [anchor]
    while pending:
        current = pending.pop()
        for child_id in children.get(current, []):
            if child_id not in scope:
                scope.add(child_id)
                pending.append(child_id)
    return scope


def ensure_org_in_managed_scope(session, actor: User, org_id: int | None) -> None:
    """校验目标机构在操作者管理范围内。

    org_id 为 None 表示顶级机构（或无部门用户）：仅全局管理员可操作。
    """
    scope = managed_scope_ids(session, actor)
    if scope is None:
        return
    if org_id is None or org_id not in scope:
        raise ForbiddenError("目标超出管理范围")


def ensure_user_scope_in_managed_scope(
    session,
    actor: User,
    *,
    role_code: str,
    effective_role: Role | None,
    managed_org_id: int | None,
    organization_id: int | None,
) -> None:
    """防提权（契约1.30修订）：目标用户最终生效管理范围不得超出操作者。

    仅当目标为 ADMIN 且生效角色具备受锚点约束的管理位
    （机构/用户编辑位）时校验——无管理位的角色（含默认角色）不构成提权。
    目标生效锚点按「角色锚点优先，角色 SELF 回落用户级三态」解析：
    解析为全局（角色锚点或用户级为 NULL）或范围外机构即拒绝；
    用户级 SELF 收敛到目标所在机构子树（其机构已由既有校验限制在
    操作者范围内，树传递性恒安全）。
    """
    scope = managed_scope_ids(session, actor)
    if scope is None or role_code != RoleCode.ADMIN.value:
        return
    if effective_role is None or not (
        effective_role.manage_org_edit or effective_role.manage_user_edit
    ):
        return
    anchor = effective_role.managed_org_id
    if anchor is None:
        raise ForbiddenError("目标超出管理范围")
    if anchor == MANAGED_ORG_SELF_ANCHOR:
        if managed_org_id is None:
            raise ForbiddenError("目标超出管理范围")
        if managed_org_id == MANAGED_ORG_SELF_ANCHOR:
            if organization_id is None:
                return
            anchor = organization_id
        else:
            anchor = managed_org_id
    if anchor not in scope:
        raise ForbiddenError("目标超出管理范围")