Dilshod Rasulov
d0badd0a2e
Roles are stored in the new resource:action format and any legacy keys are converted once at startup by the RBAC v2 DB migration (migrate_rbac_v2), so the permission engine no longer needs to resolve legacy codes on every check. - get_user_permissions: store role permission codes as-is (no dual add). - has_permission: match new-format codes only; drop resolve_permission and the dotted-code fallbacks. - Remove now-unused resolve_permission import in middleware. - LEGACY_PERMISSION_MAP / convert_* are retained as one-shot DB-migration utilities (upgrade path), not a runtime bridge. - Tests: adapt fixtures to new-format role permissions; assert legacy keys are no longer honored at request time. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
236 lines
7.7 KiB
Python
236 lines
7.7 KiB
Python
"""
|
|
Permission Engine - Central authorization logic
|
|
Handles all permission checks with scope awareness
|
|
"""
|
|
|
|
from typing import Optional, List, Set, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException, status
|
|
from loguru import logger
|
|
|
|
from .permissions import (
|
|
Permission,
|
|
PermissionRegistry,
|
|
PERMISSIONS,
|
|
ScopeType,
|
|
)
|
|
|
|
|
|
class PermissionEngine:
|
|
"""
|
|
Central permission checking engine.
|
|
All authorization decisions go through this class.
|
|
"""
|
|
|
|
@staticmethod
|
|
def get_user_permissions(user) -> Set[str]:
|
|
"""
|
|
Get all effective permissions for a user.
|
|
Returns set of permission codes.
|
|
"""
|
|
permissions = set()
|
|
|
|
# Check if user is admin (has all permissions)
|
|
if getattr(user, 'is_admin', False):
|
|
return {p.code for p in PERMISSIONS.all()}
|
|
|
|
# Get permissions from role. Roles are stored in the new
|
|
# ``resource:action`` format (legacy keys are converted once at startup
|
|
# by the RBAC v2 DB migration), so no runtime resolution is needed.
|
|
if user.role and user.role.permissions:
|
|
for perm_code, enabled in user.role.permissions.items():
|
|
if enabled:
|
|
permissions.add(perm_code)
|
|
|
|
return permissions
|
|
|
|
@staticmethod
|
|
def has_permission(
|
|
user,
|
|
permission: str,
|
|
scope: ScopeType = ScopeType.GLOBAL,
|
|
resource_id: Optional[int] = None,
|
|
organization_id: Optional[int] = None,
|
|
workspace_id: Optional[int] = None,
|
|
) -> bool:
|
|
"""
|
|
Check if user has specific permission.
|
|
|
|
Args:
|
|
user: User object
|
|
permission: Permission code (new or legacy format)
|
|
scope: Permission scope level
|
|
resource_id: Specific resource ID (for instance scope)
|
|
organization_id: Organization context
|
|
workspace_id: Workspace context
|
|
|
|
Returns:
|
|
True if user has permission, False otherwise
|
|
"""
|
|
# Admin bypass - full access
|
|
if getattr(user, 'is_admin', False):
|
|
return True
|
|
|
|
# Get user permissions
|
|
user_permissions = PermissionEngine.get_user_permissions(user)
|
|
|
|
# Check direct permission (new resource:action format)
|
|
if permission in user_permissions:
|
|
return True
|
|
|
|
# Check wildcard permissions (e.g., vm:* grants all vm permissions)
|
|
resource = permission.split(':')[0]
|
|
wildcard = f"{resource}:*"
|
|
if wildcard in user_permissions:
|
|
return True
|
|
|
|
# Check manage permission (implies most actions)
|
|
manage_perm = f"{resource}:manage"
|
|
if manage_perm in user_permissions:
|
|
# manage implies view, update, and some actions
|
|
action = permission.split(':')[1] if ':' in permission else None
|
|
implied_actions = {'view', 'list', 'update', 'start', 'stop', 'restart'}
|
|
if action in implied_actions:
|
|
return True
|
|
|
|
return False
|
|
|
|
@staticmethod
|
|
def check_permission(
|
|
user,
|
|
permission: str,
|
|
scope: ScopeType = ScopeType.GLOBAL,
|
|
resource_id: Optional[int] = None,
|
|
organization_id: Optional[int] = None,
|
|
workspace_id: Optional[int] = None,
|
|
) -> None:
|
|
"""
|
|
Check permission and raise 403 if denied.
|
|
"""
|
|
if not PermissionEngine.has_permission(
|
|
user, permission, scope, resource_id, organization_id, workspace_id
|
|
):
|
|
logger.warning(
|
|
f"Permission denied: user={user.username}, "
|
|
f"permission={permission}, scope={scope}"
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Permission denied: {permission}"
|
|
)
|
|
|
|
@staticmethod
|
|
def has_any_permission(user, permissions: List[str]) -> bool:
|
|
"""Check if user has any of the listed permissions"""
|
|
for perm in permissions:
|
|
if PermissionEngine.has_permission(user, perm):
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def has_all_permissions(user, permissions: List[str]) -> bool:
|
|
"""Check if user has all of the listed permissions"""
|
|
for perm in permissions:
|
|
if not PermissionEngine.has_permission(user, perm):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def filter_permissions(user, permissions: List[str]) -> List[str]:
|
|
"""Filter list to only permissions user has"""
|
|
return [p for p in permissions if PermissionEngine.has_permission(user, p)]
|
|
|
|
@staticmethod
|
|
def get_effective_permissions(user) -> Dict[str, Any]:
|
|
"""
|
|
Get effective permissions with metadata.
|
|
Returns dict suitable for UI display.
|
|
"""
|
|
user_perms = PermissionEngine.get_user_permissions(user)
|
|
result = {}
|
|
|
|
for perm in PERMISSIONS.all():
|
|
has_perm = perm.code in user_perms
|
|
result[perm.code] = {
|
|
"code": perm.code,
|
|
"display_name": perm.display_name,
|
|
"description": perm.description,
|
|
"category": perm.category,
|
|
"granted": has_perm,
|
|
"source": "admin" if user.is_admin else ("role" if has_perm else None)
|
|
}
|
|
|
|
return result
|
|
|
|
@staticmethod
|
|
def get_permissions_by_category(user) -> Dict[str, List[Dict]]:
|
|
"""Get effective permissions grouped by category"""
|
|
effective = PermissionEngine.get_effective_permissions(user)
|
|
by_category = {}
|
|
|
|
for perm_info in effective.values():
|
|
category = perm_info["category"]
|
|
if category not in by_category:
|
|
by_category[category] = []
|
|
by_category[category].append(perm_info)
|
|
|
|
return by_category
|
|
|
|
@staticmethod
|
|
def can_assign_permission(assigner, permission: str) -> bool:
|
|
"""
|
|
Check if user can assign a specific permission to others.
|
|
Users cannot grant permissions they don't have themselves.
|
|
"""
|
|
# Must have role management permission
|
|
if not PermissionEngine.has_permission(assigner, "role:manage"):
|
|
return False
|
|
|
|
# Cannot assign permission you don't have (unless admin)
|
|
if not assigner.is_admin:
|
|
if not PermissionEngine.has_permission(assigner, permission):
|
|
return False
|
|
|
|
return True
|
|
|
|
@staticmethod
|
|
def validate_role_permissions(assigner, permissions: Dict[str, bool]) -> tuple[Dict, List[str]]:
|
|
"""
|
|
Validate permissions that user is trying to assign.
|
|
Returns (valid_permissions, denied_permissions)
|
|
"""
|
|
valid = {}
|
|
denied = []
|
|
|
|
for perm, enabled in permissions.items():
|
|
if enabled:
|
|
if PermissionEngine.can_assign_permission(assigner, perm):
|
|
valid[perm] = True
|
|
else:
|
|
denied.append(perm)
|
|
else:
|
|
valid[perm] = False
|
|
|
|
return valid, denied
|
|
|
|
|
|
# Convenience functions
|
|
def require_permission(
|
|
user,
|
|
permission: str,
|
|
scope: ScopeType = ScopeType.GLOBAL,
|
|
resource_id: Optional[int] = None,
|
|
) -> None:
|
|
"""Require permission or raise 403"""
|
|
PermissionEngine.check_permission(user, permission, scope, resource_id)
|
|
|
|
|
|
def check_permission(user, permission: str) -> bool:
|
|
"""Check if user has permission"""
|
|
return PermissionEngine.has_permission(user, permission)
|
|
|
|
|
|
def get_user_permissions(user) -> Set[str]:
|
|
"""Get all user permissions"""
|
|
return PermissionEngine.get_user_permissions(user)
|