server-core/app/rbac.py

# -*- coding: utf-8 -*-
"""接口层 RBAC 守卫工厂：角色 / 权限 / 数据范围守卫 + 审计写入。

判定语义在领域层 ``domain/rules.py``；本文件只把规则映射为 HTTP 异常。
"""
from __future__ import annotations

from collections.abc import Callable

from fastapi import Depends, HTTPException, Request

from .api.dependencies import get_current_user
from .domain.rules import (
    is_government,
    permission_granted,
    region_in_scope,
    role_allowed,
    scope_level_ok,
    sub_role_allowed,
)
from .infrastructure.repositories import Database


def require_roles(*roles: str) -> Callable:
    """要求当前用户业务角色 ∈ roles；否则 403。"""
    def dep(user: dict = Depends(get_current_user)) -> dict:
        if not role_allowed(user, *roles):
            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。"""
    def dep(user: dict = Depends(get_current_user)) -> dict:
        if not sub_role_allowed(user, *sub_roles):
            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 not permission_granted(user, perm):
            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 not is_government(user):
            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 not scope_level_ok(user, 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_in_scope(user, region_id):
            raise HTTPException(status_code=403, detail="Forbidden: out of data scope")
        return user
    return dep


async 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 ""
    await db.audit.add(
        action=action,
        resource=resource,
        resource_id=resource_id,
        detail=detail,
        ip=ip,
        user_id=(user or {}).get("id"),
    )