103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
from fastapi import Depends, Request
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
from app.core.security import STATE_ACCESS_TOKEN, get_current_user
|
|
from shared.exceptions import ForbiddenError
|
|
from shared.tenant import STATE_TENANT_ID
|
|
|
|
_ADMINS = {"platform_admin", "tenant_owner", "tenant_admin"}
|
|
_EDITORS = {"tenant_editor"}
|
|
_READ_ONLY = {"tenant_viewer"}
|
|
|
|
|
|
def user_has_permission(user, permission: str) -> bool:
|
|
roles = set(user.roles)
|
|
if roles & _ADMINS or "workspace.*" in roles:
|
|
return True
|
|
if permission in roles:
|
|
return True
|
|
parts = permission.split(".")
|
|
return len(parts) >= 3 and f"{parts[0]}.{parts[1]}.manage" in roles
|
|
|
|
|
|
def _roles_allow(roles: set[str], permission: str) -> bool:
|
|
if roles & _ADMINS or "workspace.*" in roles:
|
|
return True
|
|
if roles & _EDITORS:
|
|
return True
|
|
if roles & _READ_ONLY:
|
|
return permission.endswith(".read") or permission.endswith(".enabled")
|
|
if permission in roles:
|
|
return True
|
|
parts = permission.split(".")
|
|
return len(parts) >= 3 and f"{parts[0]}.{parts[1]}.manage" in roles
|
|
|
|
|
|
async def _core_membership_roles(request: Request) -> set[str]:
|
|
cached = getattr(request.state, "workspace_core_roles", None)
|
|
if cached is not None:
|
|
return cached
|
|
token = getattr(request.state, STATE_ACCESS_TOKEN, None)
|
|
tenant_id = getattr(request.state, STATE_TENANT_ID, None)
|
|
if not token or not tenant_id:
|
|
request.state.workspace_core_roles = set()
|
|
return set()
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
response = await client.get(
|
|
f"{settings.core_service_url.rstrip('/')}/api/v1/me",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
except Exception:
|
|
request.state.workspace_core_roles = set()
|
|
return set()
|
|
if response.status_code != 200:
|
|
request.state.workspace_core_roles = set()
|
|
return set()
|
|
data = response.json()
|
|
roles: set[str] = set()
|
|
if data.get("platform_role"):
|
|
roles.add(str(data["platform_role"]))
|
|
for membership in data.get("memberships") or []:
|
|
if str(membership.get("tenant_id")) != str(tenant_id):
|
|
continue
|
|
if str(membership.get("status") or "active") not in {"active", "ACTIVE"}:
|
|
continue
|
|
if membership.get("role"):
|
|
roles.add(str(membership["role"]))
|
|
if membership.get("is_owner"):
|
|
roles.add("tenant_owner")
|
|
request.state.workspace_core_roles = roles
|
|
return roles
|
|
|
|
|
|
def require_permissions(*permissions: str):
|
|
async def dependency(request: Request, user=Depends(get_current_user)):
|
|
if not settings.auth_required:
|
|
return user
|
|
if any(user_has_permission(user, permission) for permission in permissions):
|
|
return user
|
|
core_roles = await _core_membership_roles(request)
|
|
if any(_roles_allow(core_roles, permission) for permission in permissions):
|
|
return user
|
|
raise ForbiddenError(
|
|
"این عملیات برای نقش فعلی فضای کاری مجاز نیست.",
|
|
error_code="permission_denied",
|
|
details={"required": list(permissions)},
|
|
)
|
|
|
|
return dependency
|
|
|
|
|
|
async def require_workspace_access(request: Request, user=Depends(get_current_user)):
|
|
if not settings.auth_required:
|
|
return user
|
|
roles = set(user.roles)
|
|
if roles & _ADMINS or any(role.startswith("workspace.") for role in roles):
|
|
return user
|
|
core_roles = await _core_membership_roles(request)
|
|
if core_roles & (_ADMINS | _EDITORS | _READ_ONLY):
|
|
return user
|
|
raise ForbiddenError("دسترسی به فضای کاری برای این سازمان مجاز نیست.", error_code="permission_denied")
|