Cheng Zhou
278 lines
9.3 KiB
Python
278 lines
9.3 KiB
Python
"""接口级权限管理API"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.deps import get_current_user, get_db_session, is_system_admin, require_study_member, require_system_permission
|
||
from app.core.project_permissions import (
|
||
get_api_endpoint_permissions,
|
||
replace_api_endpoint_permissions,
|
||
get_missing_prerequisites,
|
||
)
|
||
from app.crud import member as member_crud
|
||
from app.models.api_endpoint_registry import ApiEndpointRegistry
|
||
from app.models.study import Study
|
||
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, PROJECT_PERMISSION_ROLES, OPERATION_PREREQUISITES
|
||
|
||
router = APIRouter(prefix="/api-permissions", tags=["api-permissions"])
|
||
|
||
# 项目级路由（需要 study_id，挂载到 /studies/{study_id}/api-permissions）
|
||
study_router = APIRouter(prefix="/api-permissions", tags=["api-permissions"])
|
||
|
||
|
||
async def _get_configurable_roles(db: AsyncSession, study_id: uuid.UUID) -> list[str]:
|
||
result = await db.execute(select(Study).where(Study.id == study_id))
|
||
study = result.scalar_one_or_none()
|
||
active_roles = [role for role in (study.active_roles if study else []) if isinstance(role, str) and role.strip()]
|
||
return list(dict.fromkeys([*PROJECT_PERMISSION_ROLES, *active_roles]))
|
||
|
||
|
||
@router.get(
|
||
"/operations",
|
||
summary="获取系统中所有权限操作",
|
||
description="返回系统中所有权限操作及其描述和前置权限",
|
||
)
|
||
async def list_api_operations() -> dict[str, list[dict]]:
|
||
"""获取所有权限操作"""
|
||
operations_list = [
|
||
{
|
||
"operation_key": key,
|
||
"module": config["module"],
|
||
"action": config["action"],
|
||
"description": config["description"],
|
||
"default_roles": config["default_roles"],
|
||
"prerequisite_permissions": config.get("prerequisite_permissions", []),
|
||
}
|
||
for key, config in API_ENDPOINT_PERMISSIONS.items()
|
||
]
|
||
|
||
return {"operations": operations_list}
|
||
|
||
|
||
@router.get(
|
||
"/endpoints",
|
||
summary="获取系统中所有已注册的API端点",
|
||
description="返回系统中所有已注册的API端点及其元数据",
|
||
)
|
||
async def list_api_endpoints(
|
||
db: Annotated[AsyncSession, Depends(get_db_session)],
|
||
) -> dict[str, list[dict]]:
|
||
"""获取所有已注册的API端点"""
|
||
result = await db.execute(select(ApiEndpointRegistry))
|
||
endpoints = result.scalars().all()
|
||
|
||
endpoints_list = [
|
||
{
|
||
"endpoint_key": ep.endpoint_key,
|
||
"method": ep.method,
|
||
"path": ep.path,
|
||
"module": ep.module,
|
||
"action": ep.action,
|
||
"description": ep.description,
|
||
"default_roles": ep.default_roles.split(",") if ep.default_roles else [],
|
||
}
|
||
for ep in endpoints
|
||
]
|
||
|
||
return {"endpoints": endpoints_list}
|
||
|
||
|
||
@router.get(
|
||
"/operations/prerequisites",
|
||
summary="获取所有权限操作的前置权限依赖",
|
||
description="返回系统中所有权限操作及其前置权限依赖关系",
|
||
)
|
||
async def list_operation_prerequisites() -> dict[str, dict]:
|
||
"""获取所有权限操作的前置权限依赖"""
|
||
prerequisites_map = {}
|
||
for operation_key, prerequisites in OPERATION_PREREQUISITES.items():
|
||
prerequisites_map[operation_key] = {
|
||
"prerequisites": prerequisites,
|
||
"description": API_ENDPOINT_PERMISSIONS.get(operation_key, {}).get("description", ""),
|
||
}
|
||
|
||
return {"prerequisites": prerequisites_map}
|
||
|
||
|
||
@router.get(
|
||
"/{endpoint_key}/prerequisites",
|
||
summary="获取特定操作的缺失前置权限",
|
||
description="检查指定角色对特定操作的前置权限是否满足",
|
||
)
|
||
async def check_operation_prerequisites(
|
||
study_id: uuid.UUID,
|
||
endpoint_key: str,
|
||
role: str,
|
||
db: Annotated[AsyncSession, Depends(get_db_session)],
|
||
) -> dict:
|
||
"""检查特定操作的前置权限
|
||
|
||
返回格式:
|
||
{
|
||
"endpoint_key": "subjects:create",
|
||
"role": "CRA",
|
||
"has_main_permission": true,
|
||
"prerequisites": ["sites:read"],
|
||
"missing_prerequisites": [],
|
||
"can_perform": true
|
||
}
|
||
"""
|
||
from app.core.project_permissions import role_has_api_permission
|
||
|
||
if role == "ADMIN":
|
||
return {
|
||
"endpoint_key": endpoint_key,
|
||
"role": role,
|
||
"has_main_permission": True,
|
||
"prerequisites": OPERATION_PREREQUISITES.get(endpoint_key, []),
|
||
"missing_prerequisites": [],
|
||
"can_perform": True,
|
||
}
|
||
|
||
has_main = await role_has_api_permission(
|
||
db, study_id, role, endpoint_key, check_prerequisites=False
|
||
)
|
||
missing = await get_missing_prerequisites(db, study_id, role, endpoint_key)
|
||
|
||
return {
|
||
"endpoint_key": endpoint_key,
|
||
"role": role,
|
||
"has_main_permission": has_main,
|
||
"prerequisites": OPERATION_PREREQUISITES.get(endpoint_key, []),
|
||
"missing_prerequisites": missing,
|
||
"can_perform": has_main and len(missing) == 0,
|
||
}
|
||
|
||
|
||
@study_router.get(
|
||
"/me",
|
||
summary="获取当前用户在项目内的有效接口权限",
|
||
description="返回当前用户项目角色对应的有效权限，用于前端菜单和路由判断",
|
||
response_model=None,
|
||
)
|
||
async def get_my_study_api_permissions(
|
||
study_id: uuid.UUID,
|
||
_=Depends(require_study_member()),
|
||
current_user=Depends(get_current_user),
|
||
db: Annotated[AsyncSession, Depends(get_db_session)] = None,
|
||
):
|
||
"""获取当前用户当前项目角色的有效权限。"""
|
||
if is_system_admin(current_user):
|
||
return {
|
||
"ADMIN": {
|
||
endpoint_key: {"allowed": True}
|
||
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys()
|
||
}
|
||
}
|
||
|
||
membership = await member_crud.get_member(db, study_id, current_user.id)
|
||
role = membership.role_in_study if membership and membership.is_active else ""
|
||
permissions = await get_api_endpoint_permissions(db, study_id)
|
||
role_permissions = permissions.get(role, {})
|
||
return {
|
||
role: {
|
||
endpoint_key: role_permissions.get(endpoint_key, {"allowed": False})
|
||
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys()
|
||
}
|
||
}
|
||
|
||
|
||
@study_router.get(
|
||
"",
|
||
summary="获取项目的接口级权限矩阵",
|
||
description="返回项目中各角色对API端点的权限配置",
|
||
response_model=None,
|
||
)
|
||
async def get_study_api_permissions(
|
||
study_id: uuid.UUID,
|
||
_=Depends(require_system_permission("system:permissions:project_config")),
|
||
db: Annotated[AsyncSession, Depends(get_db_session)] = None,
|
||
):
|
||
"""获取项目的接口级权限矩阵
|
||
|
||
返回格式:
|
||
{
|
||
"role": {
|
||
"endpoint_key": {
|
||
"allowed": true/false
|
||
}
|
||
}
|
||
}
|
||
"""
|
||
permissions = await get_api_endpoint_permissions(db, study_id)
|
||
|
||
# 构建返回格式（get_api_endpoint_permissions 已返回 {role: {key: {"allowed": bool}}}）
|
||
result: dict[str, dict[str, dict[str, bool]]] = {}
|
||
for role in await _get_configurable_roles(db, study_id):
|
||
if role == "ADMIN":
|
||
continue
|
||
result[role] = {}
|
||
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys():
|
||
perm = permissions.get(role, {}).get(endpoint_key, {"allowed": False})
|
||
# perm 已经是 {"allowed": bool}，直接使用
|
||
allowed = perm["allowed"] if isinstance(perm, dict) else bool(perm)
|
||
result[role][endpoint_key] = {"allowed": allowed}
|
||
|
||
return result
|
||
|
||
|
||
@study_router.put(
|
||
"",
|
||
summary="更新项目的接口级权限矩阵",
|
||
description="批量更新项目中各角色对API端点的权限配置",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
)
|
||
async def update_study_api_permissions(
|
||
study_id: uuid.UUID,
|
||
payload: dict[str, dict[str, bool]],
|
||
_=Depends(require_system_permission("system:permissions:project_config")),
|
||
current_user=Depends(get_current_user),
|
||
db: Annotated[AsyncSession, Depends(get_db_session)] = None,
|
||
):
|
||
"""更新项目的接口级权限矩阵
|
||
|
||
请求体格式:
|
||
{
|
||
"role": {
|
||
"endpoint_key": true/false
|
||
}
|
||
}
|
||
"""
|
||
# 验证输入
|
||
configurable_roles = set(await _get_configurable_roles(db, study_id))
|
||
for role in payload.keys():
|
||
if role == "ADMIN":
|
||
continue
|
||
if role not in configurable_roles:
|
||
raise ValueError(f"无效的角色: {role}")
|
||
if role == "PM" and not is_system_admin(current_user):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="仅系统管理员可修改项目负责人权限",
|
||
)
|
||
|
||
# 替换权限配置
|
||
await replace_api_endpoint_permissions(db, study_id, payload)
|
||
|
||
# 返回更新后的权限矩阵
|
||
permissions = await get_api_endpoint_permissions(db, study_id)
|
||
|
||
result: dict[str, dict[str, dict[str, bool]]] = {}
|
||
for role in await _get_configurable_roles(db, study_id):
|
||
if role == "ADMIN":
|
||
continue
|
||
result[role] = {}
|
||
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys():
|
||
perm = permissions.get(role, {}).get(endpoint_key, {"allowed": False})
|
||
allowed = perm["allowed"] if isinstance(perm, dict) else bool(perm)
|
||
result[role][endpoint_key] = {"allowed": allowed}
|
||
|
||
return result
|