94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from sqlalchemy import and_, or_, select
|
||
|
|
|
||
|
|
from app.api.deps import CurrentUserContext
|
||
|
|
from app.core.agent_asset_scope import (
|
||
|
|
AGENT_ASSET_PLATFORM_SCOPE,
|
||
|
|
AGENT_ASSET_PLATFORM_TENANT_ID,
|
||
|
|
AGENT_ASSET_TENANT_SCOPE,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def stable_user_principal(current_user: CurrentUserContext) -> str:
|
||
|
|
"""返回不可由显示名变化影响的审计主体。"""
|
||
|
|
|
||
|
|
employee_id = str(current_user.employee_id or "").strip()
|
||
|
|
if employee_id:
|
||
|
|
return f"employee:{employee_id}"
|
||
|
|
username = str(current_user.username or "").strip().casefold()
|
||
|
|
if username:
|
||
|
|
return f"username:{username}"
|
||
|
|
raise PermissionError("当前登录用户缺少稳定身份标识。")
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True, slots=True)
|
||
|
|
class AgentAssetAccessScope:
|
||
|
|
tenant_id: str
|
||
|
|
is_platform_admin: bool = False
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_user(cls, current_user: CurrentUserContext) -> AgentAssetAccessScope:
|
||
|
|
tenant_id = str(current_user.tenant_id or "").strip()
|
||
|
|
if not tenant_id or tenant_id == AGENT_ASSET_PLATFORM_TENANT_ID:
|
||
|
|
raise PermissionError("当前登录会话缺少有效租户。")
|
||
|
|
return cls(tenant_id=tenant_id, is_platform_admin=bool(current_user.is_admin))
|
||
|
|
|
||
|
|
def visibility_clause(self, model: Any) -> Any:
|
||
|
|
return or_(
|
||
|
|
and_(
|
||
|
|
model.scope == AGENT_ASSET_PLATFORM_SCOPE,
|
||
|
|
model.tenant_id == AGENT_ASSET_PLATFORM_TENANT_ID,
|
||
|
|
),
|
||
|
|
and_(
|
||
|
|
model.scope == AGENT_ASSET_TENANT_SCOPE,
|
||
|
|
model.tenant_id == self.tenant_id,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
def can_write(self, resource: Any) -> bool:
|
||
|
|
scope = str(getattr(resource, "scope", "") or "").strip()
|
||
|
|
tenant_id = str(getattr(resource, "tenant_id", "") or "").strip()
|
||
|
|
if scope == AGENT_ASSET_PLATFORM_SCOPE:
|
||
|
|
return bool(
|
||
|
|
self.is_platform_admin and tenant_id == AGENT_ASSET_PLATFORM_TENANT_ID
|
||
|
|
)
|
||
|
|
return scope == AGENT_ASSET_TENANT_SCOPE and tenant_id == self.tenant_id
|
||
|
|
|
||
|
|
def require_write(self, resource: Any) -> None:
|
||
|
|
scope = str(getattr(resource, "scope", "") or "").strip()
|
||
|
|
tenant_id = str(getattr(resource, "tenant_id", "") or "").strip()
|
||
|
|
if scope == AGENT_ASSET_PLATFORM_SCOPE:
|
||
|
|
if tenant_id != AGENT_ASSET_PLATFORM_TENANT_ID:
|
||
|
|
raise LookupError("Asset not found")
|
||
|
|
if not self.is_platform_admin:
|
||
|
|
raise PermissionError("只有平台管理员可以修改平台资产。")
|
||
|
|
return
|
||
|
|
if scope != AGENT_ASSET_TENANT_SCOPE or tenant_id != self.tenant_id:
|
||
|
|
raise LookupError("Asset not found")
|
||
|
|
|
||
|
|
|
||
|
|
def tenant_resource_identity(tenant_id: str) -> tuple[str, str]:
|
||
|
|
normalized = str(tenant_id or "").strip()
|
||
|
|
if not normalized or normalized == AGENT_ASSET_PLATFORM_TENANT_ID:
|
||
|
|
raise ValueError("tenant_id 必须是有效的企业租户。")
|
||
|
|
return normalized, AGENT_ASSET_TENANT_SCOPE
|
||
|
|
|
||
|
|
|
||
|
|
def platform_resource_identity() -> tuple[str, str]:
|
||
|
|
return AGENT_ASSET_PLATFORM_TENANT_ID, AGENT_ASSET_PLATFORM_SCOPE
|
||
|
|
|
||
|
|
|
||
|
|
def platform_asset_statement():
|
||
|
|
"""平台初始化任务只能查询平台资产，不能按 code 命中租户覆盖项。"""
|
||
|
|
|
||
|
|
from app.models.agent_asset import AgentAsset
|
||
|
|
|
||
|
|
return select(AgentAsset).where(
|
||
|
|
AgentAsset.scope == AGENT_ASSET_PLATFORM_SCOPE,
|
||
|
|
AgentAsset.tenant_id == AGENT_ASSET_PLATFORM_TENANT_ID,
|
||
|
|
)
|