TermiNexus
9be878b84a
P1 通用查询：auth/managers.py 新增 20 个查询方法（Role/Company/Department/Position/Menu/Tenant 六个 Manager）；auth/ram.py 新增 RAMManager.ListPolicyResourceUUIDs；stats/manager.py 新增 StatsManager.ListAutoResetConfigs。 P2 App 体系：models.py 新增 App/TenantApp/CompanyApp/UserApp/AppConfig 五个 managed=False 镜像模型，字段与业务侧逐一对齐。 P3 工具下沉：新增 config.py 配置注入层与 money/json_encoder/celery_utils/exception_handler/redis_lock/ip_filter/storage/datetime_utils/string_utils 九个工具模块；model_utils.UUIDObj 增加 PascalCase 别名 Str/Bytes/UUID 与模块级转换函数。 同时移除 commission/stats/CashBook 镜像模型的重复索引声明，消除 Django models.E030。
280 lines
11 KiB
Python
280 lines
11 KiB
Python
"""模块：RAM 资源访问管理（Resource Access Management）服务层
|
||
|
||
RAM 是统一"主体→资源→操作"授权模型（借鉴 AWS RAM/IAM），与现有 RBAC 融合：
|
||
|
||
主体 Principal ：Role（角色）/ User（用户）
|
||
资源 Resource ：app（应用）/ menu（菜单）/ data_scope（数据权限，预留）
|
||
操作 Action ：view（可见）/ use（使用）/ manage（管理）/ grant（授权）
|
||
作用域 Scope ：授权只在归属租户/公司内生效（null=全租户/全公司，仅 super 可授）
|
||
|
||
与现有权限体系融合：
|
||
- App 可见性 = 安装状态（AppManager.CheckAccess）∩ RAM(角色→App, view)
|
||
- 菜单可见性 = 公司过滤 ∧ 权限码校验 ∧ 应用安装 ∧ RAM(角色→Menu, view)（可选启用）
|
||
- 数据权限暂不接管（RoleScope 四级已实现），预留 ResourceType=data_scope
|
||
|
||
判定优先级（与现有体系一致）：
|
||
- 管理员（admin:super/tenant/company）→ True（豁免 RAM 校验）
|
||
- 其余按 RAMGrant 命中（GrantStatus=1 + 作用域匹配）
|
||
"""
|
||
import logging
|
||
import uuid as _uuid
|
||
|
||
from gvsdsdk.models import RAMGrant
|
||
from gvsdsdk.model_utils import UUIDObj
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# === 资源类型 ===
|
||
RESOURCE_APP = 'app' # 应用（AppUUID）
|
||
RESOURCE_MENU = 'menu' # 菜单（MenuUUID）
|
||
RESOURCE_DATA_SCOPE = 'data_scope' # 数据权限（预留，暂不接管）
|
||
|
||
# === 操作 ===
|
||
ACTION_VIEW = 'view' # 可见（菜单注入/列表可见）
|
||
ACTION_USE = 'use' # 使用（打开应用/调用 API）
|
||
ACTION_MANAGE = 'manage' # 管理（安装/配置）
|
||
ACTION_GRANT = 'grant' # 授权（分发 RAM 授权）
|
||
|
||
# === 主体类型 ===
|
||
PRINCIPAL_ROLE = 'role' # 角色
|
||
PRINCIPAL_USER = 'user' # 用户
|
||
|
||
|
||
def _ToBytes(value):
|
||
"""统一转 bytes（兼容 bytes/UUIDObj/uuid.UUID/str）"""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, bytes):
|
||
return value
|
||
if hasattr(value, 'bytes'):
|
||
return value.bytes
|
||
if isinstance(value, str):
|
||
try:
|
||
return UUIDObj(value).bytes
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
class RAMManager:
|
||
"""RAM 授权统一服务层"""
|
||
|
||
# ==================== 授权 ====================
|
||
|
||
@staticmethod
|
||
def GrantAccess(PrincipalType, PrincipalUUID, ResourceType, ResourceUUID,
|
||
Action, TenantUUID=None, CompanyUUID=None, GrantBy=None):
|
||
"""授予访问权限（幂等 upsert：已存在同组合则激活，不存在则创建）
|
||
|
||
Args:
|
||
PrincipalType: 主体类型（role/user）
|
||
PrincipalUUID: 主体 UUID（角色或用户）
|
||
ResourceType: 资源类型（app/menu/data_scope）
|
||
ResourceUUID: 资源 UUID
|
||
Action: 操作（view/use/manage/grant）
|
||
TenantUUID: 归属租户（None=跨租户授权，仅 super 可用）
|
||
CompanyUUID: 归属公司（None=租户内全部公司）
|
||
GrantBy: 授权人 UUID（可选）
|
||
|
||
Returns:
|
||
(grant_record, err)：err 非 None 表示失败
|
||
"""
|
||
principal_bytes = _ToBytes(PrincipalUUID)
|
||
resource_bytes = _ToBytes(ResourceUUID)
|
||
tenant_bytes = _ToBytes(TenantUUID)
|
||
company_bytes = _ToBytes(CompanyUUID)
|
||
grant_by_bytes = _ToBytes(GrantBy)
|
||
if not principal_bytes or not resource_bytes:
|
||
return None, '主体/资源 UUID 无效'
|
||
if PrincipalType not in (PRINCIPAL_ROLE, PRINCIPAL_USER):
|
||
return None, f'主体类型无效: {PrincipalType}'
|
||
if ResourceType not in (RESOURCE_APP, RESOURCE_MENU, RESOURCE_DATA_SCOPE):
|
||
return None, f'资源类型无效: {ResourceType}'
|
||
if Action not in (ACTION_VIEW, ACTION_USE, ACTION_MANAGE, ACTION_GRANT):
|
||
return None, f'操作无效: {Action}'
|
||
|
||
grant, created = RAMGrant.objects.get_or_create(
|
||
PrincipalType=PrincipalType,
|
||
PrincipalUUID=principal_bytes,
|
||
ResourceType=ResourceType,
|
||
ResourceUUID=resource_bytes,
|
||
Action=Action,
|
||
TenantUUID=tenant_bytes,
|
||
CompanyUUID=company_bytes,
|
||
defaults={
|
||
'GrantUUID': _uuid.uuid4().bytes,
|
||
'GrantStatus': 1,
|
||
'GrantBy': grant_by_bytes,
|
||
},
|
||
)
|
||
if not created:
|
||
if grant.GrantStatus != 1:
|
||
grant.GrantStatus = 1
|
||
if grant_by_bytes:
|
||
grant.GrantBy = grant_by_bytes
|
||
grant.save(update_fields=['GrantStatus', 'GrantBy'])
|
||
elif grant_by_bytes:
|
||
grant.GrantBy = grant_by_bytes
|
||
grant.save(update_fields=['GrantBy'])
|
||
return grant, ''
|
||
|
||
@staticmethod
|
||
def RevokeAccess(GrantUUID):
|
||
"""撤销授权（按授权 UUID）"""
|
||
grant = RAMGrant.objects.filter(GrantUUID=_ToBytes(GrantUUID)).first()
|
||
if grant is None:
|
||
return False, '授权记录不存在'
|
||
grant.GrantStatus = 0
|
||
grant.save(update_fields=['GrantStatus'])
|
||
return True, ''
|
||
|
||
@staticmethod
|
||
def RevokeByResource(ResourceType, ResourceUUID, Action=None, PrincipalUUID=None):
|
||
"""按资源撤销全部/部分授权（如卸载应用时清理 RAM 授权）"""
|
||
qs = RAMGrant.objects.filter(ResourceType=ResourceType, ResourceUUID=_ToBytes(ResourceUUID))
|
||
if Action:
|
||
qs = qs.filter(Action=Action)
|
||
if PrincipalUUID:
|
||
qs = qs.filter(PrincipalUUID=_ToBytes(PrincipalUUID))
|
||
count = qs.update(GrantStatus=0)
|
||
return count
|
||
|
||
# ==================== 校验 ====================
|
||
|
||
@staticmethod
|
||
def CheckAccess(PrincipalUUID, ResourceType, ResourceUUID, Action,
|
||
TenantUUID=None, CompanyUUID=None):
|
||
"""校验主体是否对资源拥有指定操作（GrantStatus=1 + 作用域匹配）
|
||
|
||
作用域规则：
|
||
- 授权 TenantUUID 为 None（跨租户）→ 命中
|
||
- 授权 TenantUUID 等于请求租户 → 继续匹配公司
|
||
- 公司：授权 CompanyUUID 为 None（租户内全公司）→ 命中；等于请求公司 → 命中
|
||
|
||
Returns:
|
||
bool
|
||
"""
|
||
principal_bytes = _ToBytes(PrincipalUUID)
|
||
resource_bytes = _ToBytes(ResourceUUID)
|
||
tenant_bytes = _ToBytes(TenantUUID)
|
||
company_bytes = _ToBytes(CompanyUUID)
|
||
if not principal_bytes or not resource_bytes:
|
||
return False
|
||
|
||
qs = RAMGrant.objects.filter(
|
||
PrincipalUUID=principal_bytes,
|
||
ResourceType=ResourceType,
|
||
ResourceUUID=resource_bytes,
|
||
Action=Action,
|
||
GrantStatus=1,
|
||
)
|
||
if tenant_bytes is not None:
|
||
qs = qs.filter(_TenantScopeQ(tenant_bytes))
|
||
if company_bytes is not None:
|
||
qs = qs.filter(_CompanyScopeQ(company_bytes))
|
||
return qs.exists()
|
||
|
||
# ==================== 查询 ====================
|
||
|
||
@staticmethod
|
||
def ListGrants(ResourceType=None, ResourceUUID=None, PrincipalType=None,
|
||
PrincipalUUID=None, Action=None, Status=1, TenantUUID=None,
|
||
CompanyUUID=None):
|
||
"""列出授权记录（灵活过滤，返回 dict 列表）
|
||
|
||
Returns:
|
||
list[dict]：{GrantUUID, PrincipalType, PrincipalUUID, ResourceType,
|
||
ResourceUUID, Action, TenantUUID, CompanyUUID, GrantStatus,
|
||
GrantBy, GrantTime}
|
||
"""
|
||
qs = RAMGrant.objects.all()
|
||
if Status is not None:
|
||
qs = qs.filter(GrantStatus=Status)
|
||
if ResourceType:
|
||
qs = qs.filter(ResourceType=ResourceType)
|
||
if ResourceUUID:
|
||
qs = qs.filter(ResourceUUID=_ToBytes(ResourceUUID))
|
||
if PrincipalType:
|
||
qs = qs.filter(PrincipalType=PrincipalType)
|
||
if PrincipalUUID:
|
||
qs = qs.filter(PrincipalUUID=_ToBytes(PrincipalUUID))
|
||
if Action:
|
||
qs = qs.filter(Action=Action)
|
||
if TenantUUID:
|
||
qs = qs.filter(TenantUUID=_ToBytes(TenantUUID))
|
||
if CompanyUUID:
|
||
qs = qs.filter(CompanyUUID=_ToBytes(CompanyUUID))
|
||
return [RAMManager._ToFlat(g) for g in qs]
|
||
|
||
@staticmethod
|
||
def ListResourcesForPrincipal(PrincipalUUID, ResourceType, Action,
|
||
TenantUUID=None, CompanyUUID=None):
|
||
"""列出主体可访问的资源 UUID 集合（bytes 列表）
|
||
|
||
返回满足"作用域匹配 + GrantStatus=1"的资源 UUID bytes 列表，
|
||
供菜单注入/应用可见性批量判定使用（一次查询取集合，避免 N+1）。
|
||
"""
|
||
principal_bytes = _ToBytes(PrincipalUUID)
|
||
if not principal_bytes:
|
||
return []
|
||
qs = RAMGrant.objects.filter(
|
||
PrincipalUUID=principal_bytes,
|
||
ResourceType=ResourceType,
|
||
Action=Action,
|
||
GrantStatus=1,
|
||
)
|
||
if TenantUUID is not None:
|
||
qs = qs.filter(_TenantScopeQ(_ToBytes(TenantUUID)))
|
||
if CompanyUUID is not None:
|
||
qs = qs.filter(_CompanyScopeQ(_ToBytes(CompanyUUID)))
|
||
return list(qs.values_list('ResourceUUID', flat=True))
|
||
|
||
@staticmethod
|
||
def ListPolicyResourceUUIDs(ResourceType, Action, TenantUUID=None, CompanyUUID=None):
|
||
"""列出某类资源/操作下"已配置策略"的资源 UUID 集合（bytes 列表）
|
||
|
||
与 ListResourcesForPrincipal 的区别：不限定主体，返回整体策略集合，
|
||
供应用/菜单可见性判定"是否启用了 RAM 策略"（无策略=向后兼容全员可见）。
|
||
"""
|
||
qs = RAMGrant.objects.filter(
|
||
ResourceType=ResourceType,
|
||
Action=Action,
|
||
GrantStatus=1,
|
||
)
|
||
if TenantUUID is not None:
|
||
qs = qs.filter(_TenantScopeQ(_ToBytes(TenantUUID)))
|
||
if CompanyUUID is not None:
|
||
qs = qs.filter(_CompanyScopeQ(_ToBytes(CompanyUUID)))
|
||
return list(qs.values_list('ResourceUUID', flat=True))
|
||
|
||
# ==================== 内部 ====================
|
||
|
||
@staticmethod
|
||
def _ToFlat(grant):
|
||
"""RAMGrant 实例 → dict"""
|
||
return {
|
||
'GrantUUID': UUIDObj(grant.GrantUUID).str if grant.GrantUUID else '',
|
||
'PrincipalType': grant.PrincipalType,
|
||
'PrincipalUUID': UUIDObj(grant.PrincipalUUID).str if grant.PrincipalUUID else '',
|
||
'ResourceType': grant.ResourceType,
|
||
'ResourceUUID': UUIDObj(grant.ResourceUUID).str if grant.ResourceUUID else '',
|
||
'Action': grant.Action,
|
||
'TenantUUID': UUIDObj(grant.TenantUUID).str if grant.TenantUUID else '',
|
||
'CompanyUUID': UUIDObj(grant.CompanyUUID).str if grant.CompanyUUID else '',
|
||
'GrantStatus': grant.GrantStatus,
|
||
'GrantBy': UUIDObj(grant.GrantBy).str if grant.GrantBy else '',
|
||
'GrantTime': grant.GrantTime.isoformat() if grant.GrantTime else None,
|
||
}
|
||
|
||
|
||
def _TenantScopeQ(tenant_bytes):
|
||
"""租户作用域过滤：授权跨租户（TenantUUID 空）或匹配当前租户"""
|
||
from django.db.models import Q
|
||
return Q(TenantUUID__isnull=True) | Q(TenantUUID=tenant_bytes)
|
||
|
||
|
||
def _CompanyScopeQ(company_bytes):
|
||
"""公司作用域过滤：授权租户内全公司（CompanyUUID 空）或匹配当前公司"""
|
||
from django.db.models import Q
|
||
return Q(CompanyUUID__isnull=True) | Q(CompanyUUID=company_bytes)
|