Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""RBAC 依赖工厂：角色 / 权限 / 数据范围守卫 + 审计写入。
|
|
|
|
依赖以 ``Depends(get_current_user)`` 为前置，返回已装配 role/权限/
|
|
scope_region_ids 的当前用户 dict；不满足即抛 403。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
from fastapi import Depends, HTTPException, Request
|
|
|
|
from .dependencies import get_current_user
|
|
from .repositories import Database
|
|
|
|
# 数据范围层级（越靠前权限越大）
|
|
_SCOPE_ORDER = ("province", "city", "district")
|
|
|
|
|
|
def require_roles(*roles: str) -> Callable:
|
|
"""要求当前用户业务角色 ∈ roles；否则 403。"""
|
|
allowed = set(roles)
|
|
|
|
def dep(user: dict = Depends(get_current_user)) -> dict:
|
|
if user.get("role") not in allowed:
|
|
raise HTTPException(status_code=403, detail="Forbidden: insufficient role")
|
|
return user
|
|
|
|
return dep
|
|
|
|
|
|
def require_sub_roles(*sub_roles: str) -> Callable:
|
|
"""要求当前用户子角色 ∈ sub_roles（如 op_super_admin）；否则 403。"""
|
|
allowed = set(sub_roles)
|
|
|
|
def dep(user: dict = Depends(get_current_user)) -> dict:
|
|
if user.get("sub_role") not in allowed:
|
|
raise HTTPException(status_code=403, detail="Forbidden: insufficient sub-role")
|
|
return user
|
|
|
|
return dep
|
|
|
|
|
|
def require_permission(perm: str) -> Callable:
|
|
"""要求当前用户拥有权限码 ``perm``；否则 403。"""
|
|
def dep(user: dict = Depends(get_current_user)) -> dict:
|
|
if perm not in user.get("permissions", []):
|
|
raise HTTPException(status_code=403, detail=f"Forbidden: missing permission {perm}")
|
|
return user
|
|
|
|
return dep
|
|
|
|
|
|
def require_scope(min_level: str | None = None) -> Callable:
|
|
"""政务数据范围守卫：要求用户为 government 且有区域，且不低于 min_level。"""
|
|
def dep(user: dict = Depends(get_current_user)) -> dict:
|
|
if user.get("role") != "government":
|
|
raise HTTPException(status_code=403, detail="Forbidden: not government")
|
|
if not user.get("region_id"):
|
|
raise HTTPException(status_code=403, detail="Forbidden: no region scope")
|
|
if min_level and user.get("scope_level"):
|
|
if _SCOPE_ORDER.index(user["scope_level"]) > _SCOPE_ORDER.index(min_level):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Forbidden: scope below {min_level}",
|
|
)
|
|
return user
|
|
|
|
return dep
|
|
|
|
|
|
def scope_covers(region_id: str | None) -> Callable:
|
|
"""要求 ``region_id`` 落在当前用户数据范围内；否则 403。"""
|
|
def dep(user: dict = Depends(get_current_user)) -> dict:
|
|
if not region_id:
|
|
raise HTTPException(status_code=403, detail="Forbidden: target has no region")
|
|
if region_id not in user.get("scope_region_ids", []):
|
|
raise HTTPException(status_code=403, detail="Forbidden: out of data scope")
|
|
return user
|
|
|
|
return dep
|
|
|
|
|
|
def write_audit(
|
|
db: Database,
|
|
*,
|
|
action: str,
|
|
resource: str,
|
|
resource_id: str = "",
|
|
detail: str = "",
|
|
user: dict | None = None,
|
|
request: Request | None = None,
|
|
) -> None:
|
|
"""写入一条审计日志。"""
|
|
ip = request.client.host if request is not None and request.client else ""
|
|
db.audit.add(
|
|
action=action,
|
|
resource=resource,
|
|
resource_id=resource_id,
|
|
detail=detail,
|
|
ip=ip,
|
|
user_id=(user or {}).get("id"),
|
|
)
|