opsxsql/backend/app/services/role_permissions.py

"""角色权限服务 — 基于角色的权限查询。

权限逻辑:
- admin/dba/auditor 权限由角色硬编码决定，不需要查权限组
- developer 权限由关联的权限组决定
- 工单操作权限完全由角色决定
"""

from __future__ import annotations

import json
import logging
from typing import Dict, List, Optional, Set

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.models.instance import Instance
from app.models.instance_group import InstanceGroup, user_groups

logger = logging.getLogger(__name__)

# ============================================================
# 角色权限映射（硬编码）
# ============================================================
# access_type 值:
#   "all"    — 全部实例全部库
#   "groups" — 由权限组决定
#   None     — 无权限
# ticket 值: list of allowed ticket actions
ROLE_PERMISSIONS: Dict[str, dict] = {
    "admin": {
        "readonly": "all",
        "datachange": "all",
        "ddl": "all",
        "ticket": ["create", "approve", "reject", "execute", "cancel"],
    },
    "dba": {
        "readonly": "all",
        "datachange": "all",
        "ddl": "all",
        "ticket": ["approve", "reject", "execute"],
    },
    "auditor": {
        "readonly": "all",
        "datachange": None,
        "ddl": None,
        "ticket": ["approve", "reject"],
    },
    "developer": {
        "readonly": "groups",
        "datachange": "groups",
        "ddl": "groups",
        "ticket": ["create", "cancel_own"],
    },
    "guest": {
        "readonly": None,
        "datachange": None,
        "ddl": None,
        "ticket": [],
    },
}


def _get_role_config(role: str) -> dict:
    """获取角色配置，未知角色按 guest 处理。"""
    return ROLE_PERMISSIONS.get(role, ROLE_PERMISSIONS["guest"])


# ============================================================
# Public API
# ============================================================


async def get_user_permissions(db: AsyncSession, user: User) -> dict:
    """返回用户的实际权限概览。

    Returns:
        {
            "role": "developer",
            "access": {
                "readonly": "all" | "groups" | None,
                "datachange": "all" | "groups" | None,
                "ddl": "all" | "groups" | None,
            },
            "ticket": ["create", ...],
            "groups": [  # 仅 developer 类型有详情
                {
                    "group_id": 1,
                    "group_name": "xxx",
                    "access_type": "readonly",
                    "instance_ids": [1, 2],
                    "allowed_databases": ["db1", "db2"],
                },
                ...
            ],
        }
    """
    role_config = _get_role_config(user.role)
    result = {
        "role": user.role,
        "access": {
            "readonly": role_config["readonly"],
            "datachange": role_config["datachange"],
            "ddl": role_config["ddl"],
        },
        "ticket": role_config["ticket"],
        "groups": [],
    }

    # 如果角色权限由权限组决定，查询用户关联的权限组
    has_group_access = any(
        v == "groups" for v in role_config.values() if isinstance(v, str)
    )
    if has_group_access:
        groups_detail = await _get_user_groups_detail(db, user)
        result["groups"] = groups_detail

    return result


async def can_access_database(
    db: AsyncSession,
    user: User,
    instance_id: int,
    database: str,
    access_type: str,
) -> bool:
    """判断用户是否可以对指定实例的指定库执行指定类型的操作。

    Args:
        db: 数据库会话
        user: 用户对象
        instance_id: 实例 ID
        database: 数据库名
        access_type: "readonly" / "datachange" / "ddl"

    Returns:
        True 如果允许
    """
    if access_type not in ("readonly", "datachange", "ddl"):
        return False

    role_config = _get_role_config(user.role)
    access_value = role_config.get(access_type)

    # "all" — 全部实例全部库
    if access_value == "all":
        return True

    # None — 无权限
    if access_value is None:
        return False

    # "groups" — 由权限组决定
    if access_value == "groups":
        return await _check_group_access(db, user, instance_id, database, access_type)

    return False


def can_ticket_action(user: User, action: str) -> bool:
    """判断用户是否可以执行指定工单操作（纯角色判断，不需要查 DB）。

    Args:
        user: 用户对象
        action: "create" / "approve" / "reject" / "execute" / "cancel" / "cancel_own"

    Returns:
        True 如果允许
    """
    role_config = _get_role_config(user.role)
    ticket_perms = role_config.get("ticket", [])

    # cancel_own 是一个特殊权限，caller 需要额外验证工单申请人
    return action in ticket_perms


async def get_accessible_instances(
    db: AsyncSession,
    user: User,
) -> List[Instance]:
    """获取用户可访问的实例列表。

    - admin/dba/auditor（readonly=all）: 返回所有实例
    - developer: 返回其权限组关联的实例
    """
    role_config = _get_role_config(user.role)

    # 判断是否有 "all" 权限（至少有一个 access_type 为 all）
    has_all = any(v == "all" for v in role_config.values() if isinstance(v, str))
    if has_all:
        result = await db.execute(select(Instance))
        return list(result.scalars().all())

    # developer: 查询权限组关联的实例
    group_ids = await _get_user_group_ids(db, user)
    if not group_ids:
        return []

    result = await db.execute(
        select(Instance).where(Instance.group_id.in_(group_ids))
    )
    return list(result.scalars().all())


# ============================================================
# Internal helpers
# ============================================================


async def _get_user_group_ids(db: AsyncSession, user: User) -> List[int]:
    """获取用户关联的所有权限组 ID。"""
    result = await db.execute(
        select(user_groups.c.group_id).where(user_groups.c.user_id == user.id)
    )
    return [row[0] for row in result.all()]


async def _get_user_groups_detail(db: AsyncSession, user: User) -> List[dict]:
    """获取用户关联权限组的详细信息。"""
    group_ids = await _get_user_group_ids(db, user)
    if not group_ids:
        return []

    result = await db.execute(
        select(InstanceGroup).where(InstanceGroup.id.in_(group_ids))
    )
    groups = result.scalars().all()

    details = []
    for g in groups:
        instance_ids = [inst.id for inst in (g.instances or [])]
        details.append({
            "group_id": g.id,
            "group_name": g.name,
            "access_type": g.access_type,
            "instance_ids": instance_ids,
            "allowed_databases": g.get_allowed_databases(),
        })
    return details


async def _check_group_access(
    db: AsyncSession,
    user: User,
    instance_id: int,
    database: str,
    access_type: str,
) -> bool:
    """检查用户通过权限组是否能访问指定实例+库。"""
    # 获取实例信息
    inst_result = await db.execute(
        select(Instance).where(Instance.id == instance_id)
    )
    instance = inst_result.scalar_one_or_none()
    if not instance:
        return False

    if instance.group_id is None:
        # 公开实例，任何已登录用户都可以 readonly
        return access_type == "readonly"

    # 检查用户是否属于该实例所在的权限组
    group_ids = await _get_user_group_ids(db, user)
    if instance.group_id not in group_ids:
        return False

    # 获取权限组
    group_result = await db.execute(
        select(InstanceGroup).where(InstanceGroup.id == instance.group_id)
    )
    group = group_result.scalar_one_or_none()
    if not group:
        return False

    # 检查 access_type 是否匹配
    if group.access_type != access_type:
        return False

    # 检查 allowed_databases
    allowed_dbs = group.get_allowed_databases()
    if not allowed_dbs:
        # 空列表 = 该实例所有库
        return True

    return database in allowed_dbs