Pine
9c4908fb87
- /auth/login 全端口停用（403 引导手机号/小程序扫码登录） - 扫码会话接口下发 expires_in（前端二维码倒计时） - role_allowed 叠加制身份：所有账号自带 OPC 基础身份，operator/carrier 仅业务叠加，35 处 C 端端点统一放行
60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""领域层 · RBAC 判定规则（纯函数，不依赖 HTTP/数据库）。
|
|
|
|
四层架构：领域层定义核心业务模型与规则；接口层守卫（rbac.py 的 Depends
|
|
工厂）只负责调用这些规则并映射为 HTTP 异常，不承载判定语义。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# 数据范围层级（越靠前权限越大）
|
|
SCOPE_ORDER = ("province", "city", "district")
|
|
|
|
|
|
def is_platform_admin(user: dict) -> bool:
|
|
"""平台管理方（平台端运营者）：拥有全权限，可执行全部平台管理操作。
|
|
与其它端口（投资人/OPC/政务/企业/服务商/载体）不关联。"""
|
|
return user.get("role") == "operator" or user.get("sub_role") in ("op_super_admin", "superadmin")
|
|
|
|
|
|
def role_allowed(user: dict, *roles: str) -> bool:
|
|
"""当前用户业务角色是否 ∈ 允许角色集合。
|
|
|
|
身份模型（叠加制，非割裂）：每个账号都自带「OPC 基础身份」——运营方(operator)、
|
|
载体方(carrier) 等只是在其上的业务角色叠加。因此凡要求 opc_member 的能力
|
|
（C 端通用能力：浏览/点赞/留言/报名/接单…），任意登录账号均具备。
|
|
"""
|
|
if "opc_member" in roles:
|
|
return True
|
|
return user.get("role") in set(roles)
|
|
|
|
|
|
def sub_role_allowed(user: dict, *sub_roles: str) -> bool:
|
|
"""当前用户子角色是否 ∈ 允许子角色集合（如 op_super_admin）。"""
|
|
return user.get("sub_role") in set(sub_roles)
|
|
|
|
|
|
def permission_granted(user: dict, perm: str) -> bool:
|
|
"""当前用户是否拥有权限码 perm。平台管理方(operator/op_super_admin)恒授予全权限。"""
|
|
if is_platform_admin(user):
|
|
return True
|
|
return perm in user.get("permissions", [])
|
|
|
|
|
|
def is_government(user: dict) -> bool:
|
|
return user.get("role") == "government"
|
|
|
|
|
|
def scope_level_ok(user: dict, min_level: str | None) -> bool:
|
|
"""政务数据范围层级校验：不低于 min_level。"""
|
|
if not user.get("region_id"):
|
|
return False
|
|
if not min_level or not user.get("scope_level"):
|
|
return True
|
|
cur = user["scope_level"]
|
|
return cur in SCOPE_ORDER and SCOPE_ORDER.index(cur) <= SCOPE_ORDER.index(min_level)
|
|
|
|
|
|
def region_in_scope(user: dict, region_id: str | None) -> bool:
|
|
"""目标区域是否落在当前用户数据范围内。"""
|
|
return bool(region_id) and region_id in user.get("scope_region_ids", [])
|